在Spark數據集中使用custome UDF withColumn <Row>; java.lang.String不能轉換爲org.apache.spark.sql.Row

Question

我有一個包含許多字段的JSON文件。我在java中使用spark的Dataset讀取文件。

• 星火版本2.2.0

• JAVA JDK 1.8.0_121

下面是代碼。

SparkSession spark = SparkSession 
       .builder() 
       .appName("Java Spark SQL basic example") 
       .config("spark.some.config.option", "some-value") 
       .master("local") 
       .getOrCreate(); 

Dataset<Row> df = spark.read().json("jsonfile.json"); 

我想使用withColumn函數與自定義的UDF添加一個新的列。

UDF1 someudf = new UDF1<Row,String>(){ 
     public String call(Row fin) throws Exception{ 
      String some_str = fin.getAs("String"); 
      return some_str; 
     } 
    }; 
spark.udf().register("some_udf", someudf, DataTypes.StringType); 
df.withColumn("procs", callUDF("some_udf", col("columnx"))).show(); 

當我運行上面的代碼時出現轉換錯誤。 java.lang.String中不能被強制轉換爲org.apache.spark.sql.Row

問題：

1 - 正在讀入行的數據集的唯一選擇？我可以將df轉換爲字符串的df。但我無法選擇字段。

2 - 試過但未能定義用戶定義的數據類型。我無法使用此自定義UDDatatype註冊UDF。我需要用戶定義的數據類型嗎？

3 - 和主要問題，我怎麼能從字符串投到行？日誌

以下部分被複制：

Caused by: java.lang.ClassCastException: java.lang.String cannot be cast to org.apache.spark.sql.Row 
    at Risks.readcsv$1.call(readcsv.java:1) 
    at org.apache.spark.sql.UDFRegistration$$anonfun$27.apply(UDFRegistration.scala:512) 
     ... 16 more 

Caused by: org.apache.spark.SparkException: Failed to execute user defined function($anonfun$27: (string) => string) 

您的幫助將不勝感激。

Answer 1

由於UDF將在列的數據類型（不是）上執行，因此您正在獲取該異常。考慮我們有Dataset<Row> ds其中有兩列col1和col2都是字符串類型。現在，如果我們想使用UDF將col2的值轉換爲大寫。

我們可以註冊並撥打電話UDF，如下所示。

spark.udf().register("toUpper", toUpper, DataTypes.StringType); 
ds.select(col("*"),callUDF("toUpper", col("col2"))).show(); 

或者使用withColumn

ds.withColumn("Upper",callUDF("toUpper", col("col2"))).show(); 

而且UDF應該像下面。

private static UDF1 toUpper = new UDF1<String, String>() { 
    public String call(final String str) throws Exception { 
     return str.toUpperCase(); 
    } 
};