java - Apache Spark SQL StructType 和 UDF
问题描述
Spark 1.6 / Java-7
带有新列的初始数据框
# adding new column for the UDF computation:
df = df.withColumn("TEMP_COLUMN", lit(null));
UDF 函数创建新的 StructType 并将其放入单元格的正确格式是什么?
public static DataFrame compute(SQLContext sqlContext, DataFrame df) {
sqlContext.udf().register("compute", new MyUdf(), new ArrayType(new StructType(), true));
return df.withColumn("TEMP_COLUMN", functions.callUDF("compute"));
}
class MyUdf implements UDF0<List<StructType>> {
@Override
public List<StructType> call() {
...
return ? // what must be returned here? List<StructType> or List<String> or anything else?
}
+-------------------------+
|TEMP_COLUMN |
+-------------------------+
|[A[1, 2, 3], B[4, 5, 6]] |
+-------------------------+
我想要一个包含元素数组的结构,每个元素都有几个字段。
我不明白,注册类型new ArrayType(new StructType(), true)
是否正确,并且对于 UDF 函数的返回类型是否相同List<StructType>
。
数据应该如何返回?是这样new StructType(new StructField[]{new StructField(...))
吗?
解决方案
回答我自己的问题,因为我们很幸运地找到了如何做到这一点:
假设我们有一个“复杂”的结构来满足我们的需求:
MapType CLIENTS_INFO_DATA_TYPE = DataTypes.createMapType(
DataTypes.StringType,
DataTypes.createStructType(
new StructField[] {
DataTypes.createStructField("NAME_1", DataTypes.DoubleType, false),
DataTypes.createStructField("NAME_2", DataTypes.DoubleType, false),
DataTypes.createStructField("NAME_3", DataTypes.DoubleType, false)
),
true
);
StructType COMPLEX_DATA_TYPE = DataTypes.createStructType(new StructField[] {
DataTypes.createStructField("CLIENTS_INFO", CLIENTS_INFO_DATA_TYPE, true),
DataTypes.createStructField("COMMENT", DataTypes.StringType, true)
}
它的架构:
dataFrame.printSchema()
|-- COMPLEX_DATA_TYPE: struct (nullable = true)
| |-- CLIENTS_INFO: map (nullable = true)
| | |-- key: string
| | |-- value: struct (valueContainsNull = true)
| | | |-- NAME_1: double (nullable = false)
| | | |-- NAME_2: double (nullable = false)
| | | |-- NAME_3: double (nullable = false)
| |-- COMMENT: string (nullable = true)
接下来我们必须注册使用我们的结构操作的 UDF 函数:
DataFrame compute(SQLContext sqlContext, DataFrame df) {
sqlContext.udf().register(
"computeUDF",
new MyUDF(),
COMPLEX_DATA_TYPE);
return df.withColumn("TEMP_FIELD_NAME", functions.callUDF("computeUDF", field_1.getColumn(), field_2.getColumn()));
}
最后一步是 UDF 函数本身,它返回一个Row
对象(将转换为我们的结构):
public final class MyUDF implements UDF2<Double, Double, Row> {
@Override
public Row call(Double value1, Double value2) {
Map<String, Row> clientsInfoMap = new HashMap<>();
...
for (Map.Entry<String, ClientInfo> clientInfoEntry : clientsInfo.entrySet()) {
final String client = clientInfoEntry.getKey();
final ClientInfo clientInfo = clientInfoEntry.getValue();
final Double[] clientInfoValues = {10.0, 20.0, 30.0};
Row clientInfoRow = new GenericRow(clientInfoValues);
clientsInfoMap.put(client, clientInfoRow);
}
Object[] fullClientsInfo = new Object[] {clientsInfoMap, "string-as-a-comment"};
return new GenericRow(fullClientsInfo);
}
}
现在,由于它是一个结构,我们可以通过命名来使用 theTEMP_FIELD_NAME.CLIENTS_INFO
和其他任何东西来选择。
推荐阅读
- google-sheets - 让它显示一个唯一的单元格范围,具体取决于哪个单元格为 TRUE
- python-3.x - pandas 高级逗号分割
- regex - Excel VBA - Regex.Escape 方法
- linux - 使用 noxsave 禁用 avx 指令
- reactjs - 反应本机导航打开从标题按钮绘制
- javascript - 使用枚举进行条件渲染时反应渲染不可见的组件
- angular - Visual Studio 的默认角度项目模板失败,第一次运行时未处理异常
- spring - Spring Mongo:存储库不按顺序返回日期
- c++ - 临时对象的 Clang 6 中的模板参数推导被破坏
- django - 有什么方法可以覆盖 Django 中的外键删除保护?