如何在Apache Spark中聚合数据

Question

我在3个节点上有一个分布式系统，我的数据分布在这些节点之间。例如，我有一个test.csv文件，它存在于所有3个节点上，它包含4列

row   | id,  C1, C2,  C3
----------------------
row1  | A1 , c1 , c2 ,2
row2  | A1 , c1 , c2 ,1 
row3  | A1 , c11, c2 ,1 
row4  | A2 , c1 , c2 ,1 
row5  | A2 , c1 , c2 ,1 
row6  | A2 , c11, c2 ,1 
row7  | A2 , c11, c21,1 
row8  | A3 , c1 , c2 ,1
row9  | A3 , c1 , c2 ,2
row10 | A4 , c1 , c2 ,1

我想尝试聚合上面的结果集。如何聚合由id、c1、c2和c3列设置的数据并像这样输出？

row   | id,  C1, C2,  C3
----------------------
row1  | A1 , c1 , c2 ,3
row2  | A1 , c11, c2 ,1 
row3  | A2 , c1 , c2 ,2 
row4  | A2 , c11, c2 ,1 
row5  | A2 , c11, c21,1 
row6  | A3 , c1 , c2 ,3
row7  | A4 , c1 , c2 ,1

我尝试了以下方法：

from array import array 
from datetime import datetime 
import pyspark.sql 
from pyspark.sql import Row, SQLContext, StructField, StringType,  IntegerType
schema = StructType([
    StructField("id", StringType(), False),
    StructField("C1", StringType(), False), 
    StructField("C2", StringType(), False),
    StructField("C3", IntegerType(), False)])
base_rdd = sc.textFile("/home/hduser/spark-1.1.0/Data/test.tsv").map(lambda l: 
l.split(",")
rdd = base_rdd.map(lambda x: Row(id = x[0], C1 = x[1], C2 = x[2], C3 = int(x[3])))
sqlContext = SQLContext(sc)
srdd = sqlContext.inferSchema(rdd)

Answer 1

要解决您的问题，您可以执行以下步骤。我不知道python的步骤，下面是java的步骤。我希望你能把它和蟒蛇联系起来。

1. 读取csv文件

JavaRDD<String> input = sc.textFile(args[0]);

2. 从文件创建一对rdd

   JavaPairRDD<Integer,String> pairMap = input.mapToPair( new PairFunction<String, Integer, String>() { @Override public Tuple2<Integer, String> call(String line) throws Exception { String[] s = line.split(","); String key = s[0]+'#'+s[1]+'#' +s[2];// id,c1,c2 Integer value = Integer.valueOf(s[3]) //c3
return new Tuple2<Integer,String>(key, value); } });

3. 按键缩小地图

JavaPairRDD<String,Integer> result = pairMap.reduceByKey( new Function2<Integer, Integer, Integer>() {
@Override public Integer call(Integer v1, Integer v2) throws Exception { return v1+v2; } });

4. result对象包含您的预期结果，其中键为id+'#'+c1+'#'+c2，值为聚合c3。您可以根据需要进一步使用此地图。您可以在#上标记键以返回列，并可以使用apachespark-sql插入到表中

我希望这能有所帮助。

Answer 2

首先，我建议使用"com.databricks.spark.csv"读取csv文件（运行pyspark shell时需要使用"--package"加载http://spark-packages.org/package/databricks/spark-csv)。然后使用groupBy方法：

df = (sqlContext.read
  .format('com.databricks.spark.csv')
  .option("inferSchema", "true")
  .option("header", "true")
  .load("<your_file>.csv"))
df2= df.groupBy('id', 'C1', 'C2').agg({'C3': 'sum'}).sort('id', 'C1')
df.show()
+---+---+---+---+
| id| C1| C2| C3|
+---+---+---+---+
| A1| c1| c2|  2|
| A1| c1| c2|  1|
| A1|c11| c2|  1|
| A2| c1| c2|  1|
| A2| c1| c2|  1|
| A2|c11| c2|  1|
| A2|c11|c21|  1|
| A3| c1| c2|  1|
| A3| c1| c2|  2|
| A4| c1| c2|  1|
+---+---+---+---+
df2.show()
+---+---+---+-------+
| id| C1| C2|sum(C3)|
+---+---+---+-------+
| A1| c1| c2|      3|
| A1|c11| c2|      1|
| A2| c1| c2|      2|
| A2|c11| c2|      1|
| A2|c11|c21|      1|
| A3| c1| c2|      3|
| A4| c1| c2|      1|
+---+---+---+-------+

如果标签"row"很重要，您可以稍后添加它，并将"sum（C3）"重命名为"C3"。有关更多信息，请参阅Spark Python APIhttps://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.DataFrame