java - 使用spark时如何确保属于用户的所有数据都进入同一个文件？

Question

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

我们使用 java dataframe api 来针对 Spark 2.4.0 创建作业。我无法理解最合乎逻辑的做法是什么？我是否应该对用户 ID 进行分组操作，然后以某种方式收集行，除非我达到 1000 个用户，然后滚动(如果可能的话)或者有更好的方法。非常感谢任何正确方向的帮助或提示。

更新:

按照答案中的建议后，我继续执行以下代码片段，但我仍然看到正在写入 200 个文件，而不是 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();

但是如果我使用 100，那么我会看到 100 个文件，而不是 1000。然后我点击@Alexandros分享的链接，下面的代码片段在各自的目录中生成了超过20000个文件，而且执行时间也疯狂增加。

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 - Difference between df.repartition and DataFrameWriter partitionBy?

https://spark.apache.org/docs/1.6.3/api/java/org/apache/spark/sql/DataFrame.html#coalesce(int)

更新:

为了完成这项工作

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

spark.sql.shuffle.partitions(默认值:200)。因此，它不会提供 1000 个文件，但适用于 100 个文件。要使其正常工作，您必须首先汇回 1000 个分区，这与方法 2 相同。

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

我认为上面的代码将创建一百万个或更多文件，而不是 1000 个。

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

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