問題

有時候當你讀到一個紅移表:

% 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 ()