Spark 中的分桶分化

作者：你好赵伟 | 2024-04-26 16:32:06

踩

Spark 中的分桶分化

Bucketing是 Spark 和 Hive 中用于优化任务性能的一种技术。在分桶桶（集群列）中确定数据分区并防止数据混洗。根据一个或多个分桶列的值，将数据分配给预定义数量的桶。

分桶有两个主要好处：

改进的查询性能：在连接时，我们可以在相同的分桶列上明确指定桶的数量。由于每个存储桶包含相同大小的数据，因此映射端连接的性能优于存储桶表上的非存储桶表。在 map-side join 中，左侧表存储桶将准确知道右侧存储桶包含的数据集，以便以结构良好的格式执行表联接。
改进的采样：数据已经被分成更小的块，因此采样得到了改进。
加快连接操作的性能与尽量少的性能消耗：连接操作只需要定位各个桶，非整个数据集

何时使用桶列

表大小很大（> 200G）。
该表具有高基数列，这些列经常用作过滤和/或连接键。
中等大小的表，但主要用于连接一个巨大的桶化表，桶化它仍然是有益的
排序合并连接（没有存储桶）由于随机播放而不是由于数据倾斜而变慢

如何配置存储桶列

选择高基数列作为桶列。
尽量避免数据倾斜。
至少 500 个桶（因为小桶数会导致并行执行不佳）。
排序桶是可选的，但强烈推荐。

如何在 Spark 中创建数据桶

下面是在 SparkAPI 中创建存储桶的示例。bucketBy是在 spark 中创建存储桶的函数。我们需要将桶的信息保存在某处，所以这里需要使用saveAsTable来保存桶表的元数据信息。

n 是要创建的桶数

df.write.mode(“save_mode”)
.option(“path”, “s3 path/hdfs path”) \
*.bucketBy(n, ‘col1’, ‘col2’…) *
*.sortBy(‘col1’, ’ col2’) *
.saveAsTable(‘table_name’, format=‘parquet’)
df = spark.table(‘table_name’)

CREATE TABLE
  `temp`.`dm_log_app_activityinfo_user` (`log_id` STRING, `mid` STRING,
  `app_name` STRING, `start_time` STRING, `user_name` STRING, `is_new_visitor`
  INT, `user_class` STRING, `user_group` STRING, `user_id` STRING, `user_label`
  STRING, `session_id` STRING, `dt` STRING)
  USING orc
  OPTIONS (
  `serialization.format` '1'
  )
  PARTITIONED BY (dt)
  CLUSTERED BY (log_id)
  SORTED BY (log_id)
  INTO 4000 BUCKETS;
1
2
3
4
5
6
7
8
9
10
11
12
13

在上面的示例中，我们使用了 bucketBy 和 sortBy，因为在某些情况下我们有多个连接键，并且希望将整数键放在 bucketBy 中，将字符串键放在 sortBy 中。当我们做数据桶时，sortBy 是可选的。
可以根据数据大小和我们对数据运行的查询来决定存储桶大小的数量。通常，每个存储桶可能更喜欢 100 MB 到 200 MB。
存储桶表将使用以下命名约定将表保存在路径中。

如何在 Spark 上启用分桶？

默认情况下启用分桶。
或者，您可以在 Spark Shell 或属性文件中设置以下属性。