使用 spark.catalog.createTable 创建表

Question

我正在尝试使用spark.catalog.createTable 创建一个表。它需要有一个名为“id”的分区列。

基于 Scala 中的How can I use "spark.catalog.createTable" function to create a partitioned table?，我尝试了：

df = spark.range(10).withColumn("foo", F.lit("bar"))

spark.catalog.createTable("default.test_partition", schema=df.schema, **{"partitionColumnNames":"id"})

但它不起作用。 它在 Hive 中创建一个具有这些属性的表：

CREATE TABLE default.test_partition (   id BIGINT,   foo STRING ) 
WITH SERDEPROPERTIES ('partitionColumnNames'='id' ...

表的DDL实际上应该是：

CREATE TABLE default.test_partition (   foo STRING ) 
PARTITIONED BY (   id BIGINT ) 
WITH SERDEPROPERTIES (...

方法的签名是：

Signature: spark.catalog.createTable(tableName, path=None, source=None, schema=None, **options)

所以，我相信**options 中有一个特殊的参数来创建分区，但我尝试了“partitionColumnNames”、“partitionBy”、“partition”......它们都不起作用。 你知道什么是正确的关键字吗？

---

编辑： 如果您想知道我为什么要使用这种方法，有两个原因：

1. 出于个人好奇心，想知道什么可行或不可行
2. 我想在 spark 中使用“动态分区”，这需要我使用insertInto 方法（参见Overwrite specific partitions in spark dataframe write method）。但是这种方法需要先创建表，我想用spark.catalog.createTable 执行的操作，因为它看起来是正确的。

Answer 1

查看 spark.catalog here 的源代码，看起来关键字参数 options 是 schema 的替代品，并且仅在未传递 schema 参数时使用。这可以在下面看到：

"Optionally, a schema can be provided as the schema of the returned" #for options

if path is not None:
            options["path"] = path
            ...................
            ...........

        if schema is None: #this line and the line below
            df = self._jcatalog.createTable(tableName, source, description, options)
        else:
            if not isinstance(schema, StructType):
                raise TypeError("schema should be StructType")
            scala_datatype = self._jsparkSession.parseDataType(schema.json())
            df = self._jcatalog.createTable(
                tableName, source, scala_datatype, description, options)
        return DataFrame(df, self._sparkSession._wrapped)

但是，如果您希望创建一个自动化的 DDL 流程，以下函数的内容可能会对您有所帮助：

def mycreateTable(tablename,schema,partitioncols):
  schema_json = schema.json()
  ddlstring = (spark.sparkContext._jvm.org.apache.spark.sql.types.
                    DataType.fromJson(schema_json).toDDL())
  #if you dont want to DROP the table when it exists change the below line
  spark.sql(f"""DROP TABLE IF EXISTS {tablename} ;""") 

  spark.sql(f"""
  CREATE TABLE {tablename} ({ddlstring}) partitioned by ({','.join(partitioncols)}) """)

---

现在执行以下应该可以了：

mycreateTable("default.test_partition",df.schema,['id'])