Pyspark将数据从databricks写入azure sql：ValueError：推断后无法确定某些类型答案

【问题标题】：Pyspark writing data from databricks into azure sql: ValueError: Some of types cannot be determined after inferringPyspark将数据从databricks写入azure sql：ValueError：推断后无法确定某些类型
【发布时间】：2021-03-03 05:49:18
【问题描述】：

我正在使用 pyspark 将数据从 azure databricks 写入 azure sql。没有空值的代码运行良好，但是当数据框包含空值时，我收到以下错误：

databricks/spark/python/pyspark/sql/pandas/conversion.py:300: UserWarning: createDataFrame attempted Arrow optimization because 'spark.sql.execution.arrow.pyspark.enabled' is set to true; however, failed by the reason below:
  Unable to convert the field Product. If this column is not necessary, you may consider dropping it or converting to primitive type before the conversion.
Context: Unsupported type in conversion from Arrow: null
Attempting non-optimization as 'spark.sql.execution.arrow.pyspark.fallback.enabled' is set to true.
  warnings.warn(msg)

ValueError: Some of types cannot be determined after inferring

必须将数据帧写入 sql，包括空值。我该如何解决？

sqlContext = SQLContext(sc)

def to_sql(df, table):
  finaldf = sqlContext.createDataFrame(df)
  finaldf.write.jdbc(url=url, table= table, mode ="overwrite", properties = properties)

 to_sql(data, f"TF_{table.upper()}")

编辑：

解决了它创建一个将 pandas dtypes 映射到 sql dtypes 并将列和 dtypes 输出为一个字符串的函数。

def convert_dtype(df):
    df_mssql = {'int64': 'bigint', 'object': 'varchar(200)', 'float64': 'float'}
    mydict = {}
    for col in df.columns:
        if str(df.dtypes[col]) in df_mssql:
            mydict[col] = df_mssql.get(str(df.dtypes[col]))
    l = " ".join([str(k[0] + " " + k[1] + ",") for k in list(mydict.items())])
    return l[:-1]

将此字符串传递给createTableColumnTypes 选项解决了这种情况

jdbcDF.write \
    .option("createTableColumnTypes", convert_dtype(df) \
    .jdbc("jdbc:postgresql:dbserver", "schema.tablename",
          properties={"user": "username", "password": "password"})

【问题讨论】：

标签： pyspark azure-databricks apache-arrow

【解决方案1】：

为此，您需要在 write 语句中指定架构。这是文档中的一个示例，链接如下：

jdbcDF.write \
    .option("createTableColumnTypes", "name CHAR(64), comments VARCHAR(1024)") \
    .jdbc("jdbc:postgresql:dbserver", "schema.tablename",
          properties={"user": "username", "password": "password"})

https://spark.apache.org/docs/latest/sql-data-sources-jdbc.html

【讨论】：

您好，感谢您的回答。我编写了一个小函数来将 pandas dtypes 映射到一个包含列和 sql dtypes 的字符串。将在我的帖子中对此进行编辑。