如何从pyspark中的SQL中提取列名和列类型答案

【问题标题】：How to extract column name and column type from SQL in pyspark如何从pyspark中的SQL中提取列名和列类型
【发布时间】：2019-10-16 10:43:54
【问题描述】：

用于创建查询的 Spark SQL 类似于 this -

CREATE [TEMPORARY] TABLE [IF NOT EXISTS] [db_name.]table_name
  [(col_name1 col_type1 [COMMENT col_comment1], ...)]
  USING datasource
  [OPTIONS (key1=val1, key2=val2, ...)]
  [PARTITIONED BY (col_name1, col_name2, ...)]
  [CLUSTERED BY (col_name3, col_name4, ...) INTO num_buckets BUCKETS]
  [LOCATION path]
  [COMMENT table_comment]
  [TBLPROPERTIES (key1=val1, key2=val2, ...)]
  [AS select_statement]

其中[x] 表示x 是可选的。如果传递了CREATE sql 查询，我希望将输出作为以下顺序的元组-

(db_name, table_name, [(col1 name, col1 type), (col2 name, col2 type), ...])

那么有什么方法可以使用 pyspark sql 函数来做到这一点或需要正则表达式的帮助？

如果正则表达式有人可以帮忙处理正则表达式吗？

【问题讨论】：

不是直接解决您的问题，而是在查询创建后考虑过describe table_name 语句？不过，最后您需要将结果解析为所需的结构
在运行 CREATE 查询之前我想要 col 名称和类型。需要检查表是否已经存在于数据库中，或者是否具有相同的列名和类型。
列的顺序重要吗？
@thebluephantom 是的，非常。

标签： python sql apache-spark pyspark pyspark-sql

【解决方案1】：

可以通过java_gateway访问非官方API：

plan = spark_session._jsparkSession.sessionState().sqlParser().parsePlan("CREATE TABLE foobar.test (foo INT, bar STRING) USING json")
print(f"database: {plan.tableDesc().identifier().database().get()}")
print(f"table: {plan.tableDesc().identifier().table()}")
# perhaps there is a better way to convert the schemas, using JSON string hack here
print(f"schema: {StructType.fromJson(json.loads(plan.tableDesc().schema().json()))}")

输出：

database: foobar
table: test
schema: StructType(List(StructField(foo,IntegerType,true),StructField(bar,StringType,true)))

请注意，如果未定义数据库，database().get() 将失败并且应正确处理 Scala 选项。此外，如果您使用CREATE TEMPORARY VIEW，则访问器的名称会有所不同。命令可以在这里找到 https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ddl.scala#L38 https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ddl.scala#L58

【讨论】：