为什么Spark可以用bucketBy创建数千个文件？

Question

语境

Spark2.0.1,spark-submit在集群模式下。我正在从HDFS中读取一个拼花地板文件：

val spark = SparkSession.builder
      .appName("myApp")
      .config("hive.metastore.uris", "thrift://XXX.XXX.net:9083")
      .config("spark.sql.sources.bucketing.enabled", true)
      .enableHiveSupport()
      .getOrCreate()

val df = spark.read
              .format("parquet")
              .load("hdfs://XXX.XX.X.XX/myParquetFile")

我正在将DF保存到一个配置单元表中，该表包含按userid分组的50个桶：

df0.write
   .bucketBy(50, "userid")
   .saveAsTable("myHiveTable")

现在，当我查看hdfs的hive仓库/user/hive/warehouse时，有一个名为myhivetable的文件夹。它的内部是一堆part-*.parquet文件。我希望有50个文件。但是没有，有3201个文件！！！！每个分区有64个文件，为什么？对于我保存为配置单元表的不同文件，每个分区有不同数量的文件。所有的文件都很小，每个只有几十Kb！

让我补充一下，在MyParquetFile中，不同的Userid的数量大约为1,000。

为什么文件夹里有3201个文件而不是50个！它们是什么？

当我将此表读回DataFrame并打印分区数时：

val df2 = spark.sql("SELECT * FROM myHiveTable") 
println(df2.rdd.getNumPartitions)

分区的数量是正确的50，并且我确认数据是通过userid正确分区的。

问题

创建的文件数量取决于什么？

问题

有办法限制创建的文件数量吗？

问题

我应该担心这些文件吗？拥有所有这些文件是否会影响DF2的性能？人们总是说我们不应该创建太多的分区，因为这是有问题的。

问题

我发现这个信息配置单元动态分区提示，文件的数量可能与映射器的数量有关。建议在插入配置单元表时使用distribution by。我怎么能在Spark里做到呢？

问题

如果问题确实如上面的链接所示，在MapR-FS上插入数据后如何控制配置单元表的文件号，建议使用hive.merge.mapfiles或hive.merge.mapredfiles等选项合并map reduce作业后的所有小文件。在Spark中有这样的选择吗？

Answer 1

请使用spark sql，它将使用HiveContext将数据写入配置单元表，因此它将使用您在表模式中配置的桶数。

 SparkSession.builder().
  config("hive.exec.dynamic.partition", "true").
  config("hive.exec.dynamic.partition.mode", "nonstrict").
  config("hive.execution.engine","tez").
  config("hive.exec.max.dynamic.partitions","400").
  config("hive.exec.max.dynamic.partitions.pernode","400").
  config("hive.enforce.bucketing","true").
  config("optimize.sort.dynamic.partitionining","true").
  config("hive.vectorized.execution.enabled","true").
  config("hive.enforce.sorting","true").
  enableHiveSupport().getOrCreate()

spark.sql(s"insert into hiveTableName partition (partition_column) select * from  myParquetFile")

spark的bucketing实现没有满足指定的bucket大小数。每个分区都写入一个单独的文件，因此每个bucket都有很多文件。

拉维