WrapedArray的WrappedArray到java数组

我有一个类型集的列，我使用spark数据集API的collect_set()，它返回包装数组的包装数组。我想从嵌套包装数组的所有值中获取单个数组。我怎样才能做到这一点？WrapedArray的WrappedArray到java数组

例如， Cassandra表：

Col1 
{1,2,3} 
{1,5}

我正在使用Spark数据集API。
row.get(0)返回包装数组的包装数组。

来源

2017-07-26 rohanagarwal

考虑你有Dataset<Row> ds其中有value列。

+-----------------------+ 
|value     | 
+-----------------------+ 
|[WrappedArray(1, 2, 3)]| 
+-----------------------+

而且它有以下模式

root 
|-- value: array (nullable = true) 
| |-- element: array (containsNull = true) 
| | |-- element: integer (containsNull = false)

使用UDF

定义UDF1像下面。

static UDF1<WrappedArray<WrappedArray<Integer>>, List<Integer>> getValue = new UDF1<WrappedArray<WrappedArray<Integer>>, List<Integer>>() { 
     public List<Integer> call(WrappedArray<WrappedArray<Integer>> data) throws Exception { 
     List<Integer> intList = new ArrayList<Integer>(); 
     for(int i=0; i<data.size(); i++){ 
      intList.addAll(JavaConversions.seqAsJavaList(data.apply(i))); 
     } 
     return intList; 
    } 
};

注册和呼叫UDF1像下面

import static org.apache.spark.sql.functions.col; 
import static org.apache.spark.sql.functions.callUDF; 
import scala.collection.JavaConversions; 

//register UDF 
spark.udf().register("getValue", getValue, DataTypes.createArrayType(DataTypes.IntegerType)); 

//Call UDF 
Dataset<Row> ds1 = ds.select(col("*"), callUDF("getValue", col("value")).as("udf-value")); 
ds1.show();

使用爆炸功能

import static org.apache.spark.sql.functions.col; 
import static org.apache.spark.sql.functions.explode; 

Dataset<Row> ds2 = ds.select(explode(col("value")).as("explode-value")); 
ds2.show(false);

来源

2017-07-26 18:05:14 abaghel

是的，可以完成，我试过了其他我分解了这些集合，然后使用collect_set（）将它们聚合在一起，所以只有一个数组，你告诉我要爆炸collect_set（）的结果，在这两种情况下，我都有一个问题，那就是是否会有这也是我选择flatten的原因，你也可以指点我一些教程，书籍等等，用于spark + java（而不是scala）+ dataset api – rohanagarwal

我编辑了我的答案，以使用UDF获取数组。希望这可以帮助。 – abaghel

如果你有一个数据框，你可以使用UDF来flattern列表下面是简单的例子，对于DF1

import spark.implicits._ 

import org.apache.spark.sql.functions._ 
//create a dummy data 

val df = Seq(
    (1, List(1,2,3)), 
    (1, List (5,7,9)), 
    (2, List(4,5,6)), 
    (2,List(7,8,9)) 
).toDF("id", "list") 

val df1 = df.groupBy("id").agg(collect_set($"list").as("col1")) 

df1.show(false)

输出：

+---+----------------------------------------------+ 
|id |col1           | 
+---+----------------------------------------------+ 
|1 |[WrappedArray(1, 2, 3), WrappedArray(5, 7, 9)]| 
|2 |[WrappedArray(7, 8, 9), WrappedArray(4, 5, 6)]| 
+---+----------------------------------------------+ 


val testUDF = udf((list: Seq[Seq[Integer]]) => {list.flatten}) 


df1.withColumn("newCol", testUDF($"col1")).show(false)

输出

+---+----------------------------------------------+------------------+ 
|id |col1           |newCol   | 
+---+----------------------------------------------+------------------+ 
|1 |[WrappedArray(1, 2, 3), WrappedArray(5, 7, 9)]|[1, 2, 3, 5, 7, 9]| 
|2 |[WrappedArray(7, 8, 9), WrappedArray(4, 5, 6)]|[7, 8, 9, 4, 5, 6]| 
+---+----------------------------------------------+------------------+

我希望这可以帮助！

来源

2017-07-26 12:28:47

你可以请张贴相当于Java代码UDF。我在Seq >上看到了这种扁平化功能，但无法正确使用它。 – rohanagarwal

我希望这可以帮助https://stackoverflow.com/questions/35348058/how-do-i-call-a-udf-on-a-spark-dataframe-using-java –

其实我想实现扁平化，它并不像Java中的list.flatten那么简单，可能是因为Scala更丰富。文件为flatten是单行，对我来说没有意义:( – rohanagarwal

WrapedArray的WrappedArray到java数组

回答

相关问题