问题

有时候当你读到一个红移表:

% scala val original_df = spark.read。格式(“com.databricks.spark.redshift”)。选项(“url”, url)。用户选项(“用户”)。选项(“密码”,密码)。选项(“查询”,查询)。选项(“forward_spark_s3_credentials”,真正的)。选项(“tempdir”、“路径”)。load ()

火花作业将抛出一个NullPointerException:

引起的:. lang。NullPointerExceptionat org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter.write(UnsafeRowWriter.java:194)

导致

这个问题来自引发红移的读取数据的方式。亚马逊的红移数据源使用红移的卸载从红移格式读取数据:引发第一个问题卸载命令来红移,让它转储表的内容卸载格式临时文件,然后火花扫描这些临时文件。这个基于文本的卸载格式不区分一个空字符串,默认一个空字符串,都是编码为一个空字符串在结果文件中。卸载spark-redshift读取数据格式时,没有足够的信息,判断输入空字符串或一个空,,目前它只是认为这是一个零。

解决方案

在Scala中,设置可以为空来真正的对所有的字符串列:

% scala org.apache.spark.sql.types进口。{StructField, StructType, StringType} org.apache.spark.sql进口。{DataFrame, SQLContext} def setNullableStateForAllStringColumns (df: DataFrame nullable:布尔)= {StructType (df.schema。地图{案例StructField (c StringType _, m) = > StructField (c StringType nullable =可空,m)情况下StructField (c、t、n,米)= > StructField (c、t、n, m)})}

在Python中:

nullable % python def set_nullable_for_all_string_columns (df):从pyspark.sql。类型进口StructType、StructField StringType new_schema = StructType ([StructField f.name f。数据类型,可以为空,如果(isinstance (f f.metadata)。其他数据类型,StringType)) StructField (f.name f。数据类型,f。可以为空, f.metadata) for f in df.schema.fields]) return new_schema

使用这个函数,得到的模式original_df,然后修改模式字符串列可以为空从红移,然后重读:

% scala val df = spark.read。模式(setNullableStateForAllStringColumns (original_df真实))。格式(“com.databricks.spark.redshift”)。选项(“url”, url)。用户选项(“用户”)。选项(“密码”,密码)。选项(“查询”,查询)。选项(“forward_spark_s3_credentials”,真正的)。选项(“tempdir”、“路径”)。load ()