8g流量网站,百度SEO网站,跨境平台有哪些,创建网站域名多少钱目录 1.合并查询2.使用 JOIN 条件的过滤优化3.使用 Map-side Join 或 Broadcast Join4.使用 Partitioning 和 Bucketing5.利用 DataFrame API 进行优化假设 A 和 B 已经加载为 DataFramePerform left joins with specific conditions6.使用缓存或持久化7.避免笛卡尔积总结 1.合… 目录 1.合并查询2.使用 JOIN 条件的过滤优化3.使用 Map-side Join 或 Broadcast Join4.使用 Partitioning 和 Bucketing5.利用 DataFrame API 进行优化假设 A 和 B 已经加载为 DataFramePerform left joins with specific conditions6.使用缓存或持久化7.避免笛卡尔积总结 1.合并查询
如果在 SQL 中的多个 JOIN 操作是针对同一个表只是条件不同可以考虑将条件合并成一个查询从而减少对同一表的多次扫描。例如将多个 LEFT JOIN 转换成一个 JOIN使用 CASE 或 FILTER 直接处理不同的关联条件。
优化前
SELECT A.id, A.col1, A.col2, B1.col3 AS B1_col3, B2.col3 AS B2_col3
FROM A
LEFT JOIN B AS B1 ON A.id B1.id AND A.col1 B1.col4
LEFT JOIN B AS B2 ON A.id B2.id AND A.col2 B2.col4优化后
SELECT A.id, A.col1, A.col2, MAX(CASE WHEN A.col1 B.col4 THEN B.col3 END) AS B1_col3,MAX(CASE WHEN A.col2 B.col4 THEN B.col3 END) AS B2_col3
FROM A
LEFT JOIN B ON A.id B.id
GROUP BY A.id, A.col1, A.col22.使用 JOIN 条件的过滤优化
通过精简 JOIN 条件尽量减少连接的行数。例如如果 B 表中有索引列可以直接根据索引列做筛选而不依赖复杂的条件。
假设对 B 表进行的连接条件中有部分条件可以通过过滤的方式提前应用比如通过 WHERE 子句或者 JOIN 之前的 FILTER。
SELECT A.id, A.col1, A.col2, B1.col3 AS B1_col3, B2.col3 AS B2_col3
FROM A
LEFT JOIN B AS B1 ON A.id B1.id AND A.col1 B1.col4
LEFT JOIN B AS B2 ON A.id B2.id AND A.col2 B2.col4
WHERE B1.col3 IS NOT NULL OR B2.col3 IS NOT NULL3.使用 Map-side Join 或 Broadcast Join
在 Spark SQL 中当其中一个表比如 A较小且能完全加载到内存时Spark 会自动选择广播连接即将小表广播到所有工作节点进行连接计算而不是进行全表扫描。
如果你知道某个表的规模较小例如 A可以手动启用广播连接减少 shuffle 的开销。
SELECT /* BROADCAST(A) */A.id, A.col1, A.col2, B1.col3 AS B1_col3, B2.col3 AS B2_col3
FROM A
LEFT JOIN B AS B1 ON A.id B1.id AND A.col1 B1.col4
LEFT JOIN B AS B2 ON A.id B2.id AND A.col2 B2.col4在这里通过 /* BROADCAST(A) */ 强制 Spark 将 A 表广播到各个执行节点从而避免了对大表 B 进行多次 shuffle。
广播条件 A 表要相对较小可以完全加载到内存中。 B 表较大且 A 表的行数远小于 B。
4.使用 Partitioning 和 Bucketing
在分布式环境下通过合理的分区和分桶设计可以减少 JOIN 时的 shuffle 开销。尤其是对于大表可以考虑对 A 或 B 表做分区PARTITION BY或分桶BUCKET BY。 – 对 B 表进行分桶根据 id 或其他相关字段
CREATE TABLE B (id INT,col3 STRING,col4 STRING
)
USING parquet
CLUSTERED BY (id) INTO 10 BUCKETS;通过将表按某个字段进行分桶Spark 在进行连接时能够减少数据的移动和重新分配。
5.利用 DataFrame API 进行优化
如果 SQL 性能不够高可以尝试将查询转为 DataFrame API 编写Spark DataFrame API 可能在某些复杂的连接和查询场景下更加高效。
假设 A 和 B 已经加载为 DataFrame
from pyspark.sql import functions as F
Perform left joins with specific conditions
df_A spark.table(A)
df_B spark.table(B)df_B1 df_B.filter(df_B.col4.isNotNull()).select(id, col3)
df_B2 df_B.filter(df_B.col4.isNotNull()).select(id, col3)df_result df_A.join(df_B1, (df_A.id df_B1.id) (df_A.col1 df_B1.col4), left) \.join(df_B2, (df_A.id df_B2.id) (df_A.col2 df_B2.col4), left) \.select(df_A.id, df_A.col1, df_A.col2, df_B1.col3.alias(B1_col3), df_B2.col3.alias(B2_col3))df_result.show()DataFrame API 可以对复杂的 JOIN 和条件执行更多优化比如延迟执行和缓存策略。
6.使用缓存或持久化
如果你在多次查询中重复使用某些中间结果例如对 B 表的过滤结果或计算结果可以选择缓存或持久化某些 DataFrame。
df_B1_cached df_B1.cache()
df_B2_cached df_B2.cache()df_result df_A.join(df_B1_cached, (df_A.id df_B1_cached.id) (df_A.col1 df_B1_cached.col4), left) \.join(df_B2_cached, (df_A.id df_B2_cached.id) (df_A.col2 df_B2_cached.col4), left)缓存对于反复使用的子查询可以减少重新计算的开销。
7.避免笛卡尔积
笛卡尔积会导致非常高的计算开销和内存占用因此在 JOIN 时需要确保条件足够明确避免无条件的多表连接。你可以使用 EXPLAIN 来分析查询计划检查是否出现了笛卡尔积。 查询计划中的 CartesianProduct 或 CROSS JOIN
EXPLAIN SELECT A.id, A.col1, A.col2, B.col3 FROM A JOIN B ON A.id B.idSpark SQL / Hive 中查询计划可能会显示 CartesianProduct 或类似的描述指明两张表间进行了笛卡尔积连接。 Physical Plan
CartesianProduct(0)PostgreSQL、MySQL 等关系型数据库通常会标明连接类型。如果执行计划中显示了 CROSS JOIN则明确表示笛卡尔积。
- Seq Scan on table_a (cost0.00..10.00 rows100 width20)
- Seq Scan on table_b (cost0.00..10.00 rows100 width20)
- Hash Join (cost200.00..220.00 rows1000 width100)如果这里显示了 CROSS JOIN就意味着没有任何连接条件导致笛卡尔积的生成。
通过查看执行计划EXPLAIN了解是否存在不必要的全表扫描。
总结
合并查询 用 CASE WHEN 合并多个 JOIN。 简化 JOIN 条件 提前通过 WHERE 子句过滤无效数据。 广播连接 对小表使用 BROADCAST减少 shuffle 开销。 分区和分桶 对大表进行分区或分桶优化 JOIN 性能。 使用 DataFrame API 在某些复杂查询中DataFrame API 性能更优。 缓存数据 重复使用的数据可以进行缓存或持久化。 避免笛卡尔积 确保 JOIN 有明确的条件避免全表扫描。
