Delta Lake 表 PySpark/Glue 目录 EMR

Question

使用 Spark 创建数据库，位置参数指向 s3 存储桶路径，创建数据帧并使用 saveAsTable 写入增量，数据库和表都显示在指向 s3 位置的粘合点中。然后当我用

spark.sql(‘select * from mydb.tbl’)

阅读时，我得到 Table does not support reads： 但是使用时

spark.read.format(‘delta’).load(‘s3 path’)

数据返回。如何从glue 目录中查询增量表而不提供s3 路径？我认为元数据数据库存储这些信息，例如数据文件的位置。

设置。 电子病历6.15.0 Hadoop 3.3.6、Hive 3.1.4、JupyterEnterpriseGateway 2.6.0、Spark 3.4.1、Livy 0.7.1

节点通过以下分类进行配置： Spark-Hive-站点：hive.metastore.client.factory.class：com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory

delta-默认值：delta.enabled：true

代码片段：

spark = SparkSession.build.config(‘spark.jars’, ‘/usr/share/aws/delta/lib/delta-core_2.12-2.4.0.jar’).config(‘spark.sql.extension’, ‘io.delta.sql.DeltaSparkSessionExtension’).config(‘spark.sql.catalog.spark_catalog’, ‘org.apache.spark.sql.delta.catalog.DeltaCatalog’).enableHiveSupport().getOrCreate()

spark.sql(“create database mydb LOCATION ‘s3://bucket_name/mydb’”)

df.write.format(‘delta’).mode(‘overwrote’).saveAsTable(‘tbl’)

此时表和所有增量文件夹都已保存到创建数据库行存储桶中指定的 s3 中，并在 mydb/ 下使用新路径

spark.sql(‘select * from mydb.tbl’)

返回表不支持读取

spark.read.format('delta').load('path') 或 DeltaTable.forPath(spark, 'path') 是查询增量表的唯一方法，并且仍然必须提供路径，即使它已在胶水目录？

我在这里缺少什么？

Answer 1

我相信您的 Spark 会话中缺少以下配置：

"spark.hadoop.hive.metastore.client.factory.class": "com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory"

我已使用以下配置初始化我的 Spark 会话，并且能够使用

spark.sql

进行查询：

"conf": {
    "spark.sql.extensions" : "io.delta.sql.DeltaSparkSessionExtension",
    "spark.sql.catalog.spark_catalog": "org.apache.spark.sql.delta.catalog.DeltaCatalog",
    "spark.jars": "/usr/share/aws/delta/lib/delta-core.jar,/usr/share/aws/delta/lib/delta-storage.jar,/usr/share/aws/delta/lib/delta-storage-s3-dynamodb.jar",
    "spark.hadoop.hive.metastore.client.factory.class": "com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory"
}

结果：

spark.sql("select * from mydb.delta_table;").show()

+---+-------------+--------------------+
| id|creation_date|    last_update_time|
+---+-------------+--------------------+
|102|   2015-01-01|2015-01-01T13:51:...|
|101|   2015-01-01|2015-01-01T12:14:...|
|103|   2015-01-01|2015-01-01T13:51:...|
|100|   2015-01-01|2015-01-01T13:51:...|
+---+-------------+--------------------+

Answer 2

这里有一些东西你应该寻找：

1. 一旦您在 EMR 中启用了增量表，即使用

   "delta.enabled":"true"

   进行配置，您就不必提及在

   "spark.jars"

   中提到的 jar，因为一旦启用

   delta

   ，它应该已经包含在内。即使您想提及

   jars

   ，您也应该包含 AWS 文档中提到的 jar。

"/usr/share/aws/delta/lib/delta-core.jar,/usr/share/aws/delta/lib/delta-storage.jar,/usr/share/aws/delta/lib/delta-storage-s3-dynamodb.jar"

2. 由于

   delta

   表是

   v2

   表格式，您可以使用

   spark.table("<tablename>")

   从任何OFT表中读取数据，即

   Delta

   、

   Iceberg

   和

   Hudi

   。

ddf = spark.table("db.delta_table")
ddf.show()

3. 如果你想使用

   DeltaTable

   pyspark API，你也可以使用

   DeltaTable.forName()

   ，这可以让你通过表名读取表。

from delta.tables import DeltaTable
delta_tbl = DeltaTable.forName(spark, 'db.delta_table')

首先，我建议从

delta-core

 中移除 

spark.jars

罐子