我想按鍵減少數據幀。reduce 邏輯非常復雜，需要更新大約 10-15 個字段。這就是為什么我想將 DataFrame 轉換為 DataSet 并減少 Java POJO。問題問題是，在groupByKey-reduceByKey我得到一些非常奇怪的值之后。Encoders.bean(Entity.class)讀取正確的數據。請參閱代碼示例部分。變通方法替換Encoders.bean為Encoders.kryo不起作用，異常：Try to map struct<broker_name:string,server_name:string,order:int,storages:array<struct<timestamp:timestamp,storage:double>>> to Tuple1, but failed as the number of fields does not line up.我也看到了這個 workarround，但Encoders.product需要TypeTag. 我不知道如何TypeTag在 Java 代碼中創建。
1 回答
泛舟湖上清波郎朗
TA貢獻1818條經驗 獲得超3個贊
這是因為反序列化使用由 推斷出的架構上的結構匹配Encoder，并且由于 bean 類沒有自然結構，架構的字段按名稱排序。
所以如果你定義一個像你的 bean 類Entity，從 bean 推斷的模式Encoder將是
Encoders.bean(Storage.class).schema().printTreeString();
root
|-- storage: double (nullable = true)
|-- timestamp: timestamp (nullable = true)
不是
root
|-- timestamp: timestamp (nullable = true)
|-- storage: double (nullable = true)
這是應該使用的架構Dataset。換句話說，架構定義為：
StructType schema = Encoders.bean(Entity.class).schema();
或者
StructType schema = StructType.fromDDL(
"broker_name string, order integer, server_name string, " +
"storages array<struct<storage: double, timestamp: timestamp>>"
);
將是有效的，并且可以用于testData直接加載：
Dataset<Entity> ds = spark.read()
.option("multiline", "true")
.schema(schema)
.json("testData.json")
.as(Encoders.bean(Entity.class));
而您當前的架構，相當于：
StructType valid = StructType.fromDDL(
"broker_name string, order integer, server_name string, " +
"storages array<struct<timestamp: timestamp, storage: double>>"
);
不是，盡管它可以與 JSON 閱讀器一起使用，它（與 相比Encoders）按名稱匹配數據。
可以說，這種行為應該被報告為一個錯誤——直觀地說，不應該有Encoder轉儲與其自己的加載邏輯不兼容的數據的情況。
添加回答
舉報
0/150
提交
取消