在 spark 中将字符串名称转换为 sql 数据类型答案

【问题标题】：Converting string name to sql datatype in spark在 spark 中将字符串名称转换为 sql 数据类型
【发布时间】：2019-07-26 13:55:48
【问题描述】：

我一直在参考下面的文章

Spark cast column to sql type stored in string

我正在寻找 pyspark 中的等效代码。

问题是上面帖子中的答案使用了classof[DataTypes]，但是pyspark中没有DataTypes类。

我想要做的是动态创建架构。所以，我有一个如下列表：

>>> sourceToHiveTypeList
['TimestampType', 'TimestampType', 'StringType', 'StringType', 'IntegerType', 'DoubleType']

我已经定义了一个 UDF

def TableASchema(columnName, columnType): 
   return StructType([
       StructField(columnName[0], getattr(pyspark.sql.types,columnType[0]), nullable = True),
       StructField(columnName[1], getattr(pyspark.sql.types,columnType[1]), nullable = True),
       StructField(columnName[2], getattr(pyspark.sql.types,columnType[2]), nullable = True),
       StructField(columnName[3], getattr(pyspark.sql.types,columnType[3]), nullable = True),
       StructField(columnName[4], getattr(pyspark.sql.types,columnType[4]), nullable = True),
       StructField(columnName[5], getattr(pyspark.sql.types,columnType[5]), nullable = True)
      ])

当我调用上面的 UDF 时，我得到了错误：

>>> schema = TableASchema(headerColumns, sourceToHiveTypeList)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 3, in TableASchema
AttributeError: 'module' object has no attribute 'TimestampType()'

【问题讨论】：

DataTypes 不可用是什么意思？它是一个模块，而不是一个类，如果这就是你的意思...spark.apache.org/docs/2.1.0/api/python/_modules/pyspark/sql/…
@cricket_007 我相信 OP 引用了 `classOf]
在这里查看答案 - stackoverflow.com/questions/39777648/… 您可以使用普通的 scala 反射创建 DataType 的实例。当数据类型存储为 sql 数据类型字符串时，Catalyst 解析器还提供创建 DataType 实例。

标签： python apache-spark pyspark

【解决方案1】：

如果您正在寻找仅适用于原子类型的解决方案（与链接问题中的相同）：

import pyspark.sql.types

def type_for_name(s):
    return getattr(pyspark.sql.types, s)()

type_for_name("StringType")
# StringType

可以使用eval 解析复杂类型，但出于安全考虑，我会非常小心：

def type_for_name_(s):
    types = {
        t: getattr(pyspark.sql.types, t) 
        for t  in dir(pyspark.sql.types) if t.endswith("Type")}
    t = eval(s, types, {})
    return t if isinstance(t, pyspark.sql.types.DataType) else t()

type_for_name_("DecimalType(10, 2)")
# DecimalType(10,2)

一般我会推荐使用短字符串（即string、double、struct<x:integer,y:integer>，可以直接使用：

col("foo").cast("integer")

如果您需要更复杂的表示，请使用 JSON。

【讨论】：

【解决方案2】：

def toDataType(dataType: String): DataType = {
    val module =
      runtimeMirror.staticModule("org.apache.spark.sql.types." + dataType)
    runtimeMirror.reflectModule(module).instance.asInstanceOf[DataType]
}

【讨论】：