在使用Spark时，如何确保属于一个用户的所有数据都到同一个文件中？

Question

我们有一个用例来准备一个spark作业，该作业将从多个提供者读取数据，其中包含关于以任意顺序出现的用户的信息，并将它们写回S3中的文件。现在，条件是，用户的所有数据必须存在于一个文件中。大约有100万个唯一的用户，每个用户最多拥有大约10KB的数据。我们想到最多创建1000个文件，并让每个文件包含大约1000个用户的记录。

我们使用java dataframe API在spark 2.4.0上创建作业。我不知道做这件事最合乎逻辑的方法是什么？我应该在用户ID上执行group by操作，然后以某种方式收集行，除非我达到1000个用户，然后翻转（如果可能的话），还是有更好的方法。任何正确方向的帮助或提示都非常感谢。

Properties props = PropLoader.getProps("PrepareData.properties");
SparkSession spark = SparkSession.builder().appName("prepareData").master("local[*]")
    .config("fs.s3n.awsAccessKeyId", props.getProperty(Constants.S3_KEY_ID_KEY))
    .config("fs.s3n.awsSecretAccessKey", props.getProperty(Constants.S3_SECERET_ACCESS_KEY)).getOrCreate();

Dataset<Row> dataSet = spark.read().option("header", true).csv(pathToRead);
dataSet.repartition(dataSet.col("idvalue")).coalesce(1000).write().parquet(pathToWrite);

spark.close();

dataSet.repartition(1000, dataSet.col("idvalue")).write().partitionBy("idvalue").parquet(pathToWrite);

Answer 1

可以使用重新分区然后合并函数。

 Df.repartion(user_id).coalese(1000)

 Df.repartion(user_id,1000)

第一种解决方案保证不会有任何空分区，而在第二种解决方案中，有些分区可能是空的。

参考：Spark SQL-df.repartition和DataFrameWriter Partitionby之间的差异？

dataSet.repartition(dataSet.col("idvalue")).coalesce(1000).write().parquet(pathToWrite);

dataSet.repartition(1000, dataSet.col("idvalue")).write().partitionBy("idvalue").parquet(pathToWrite);

dataSet.repartition(1000, dataSet.col("idvalue")).write().parquet(pathToWrite);

它将创建1000个文件，但是在完成文件编写后，您必须通过读取每个文件来创建ids和文件之间的映射。