使用spark时如何确保属于用户的所有数据都转到同一个文件？

Question

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

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

更新：

按照答案中的建议后，我继续使用以下代码 sn-p，但我仍然看到正在写入 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，而不是 1000，那么我会看到 100 个文件。然后我点击@Alexandros 分享的链接，下面的代码 sn-p 在它们各自的目录中生成了超过 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 和文件之间的映射。