如何优化火花SQL运行它并行

Question

我是火花新手，使用有一个简单的火花申请星火SQL/hiveContext到：

1. 从蜂巢表选择数据（1个十亿行）
2. 做一些过滤，聚合包括ROW_NUMBER在窗函数通过选择第一行，组数（）和max（）等
3. 结果写入到HBase的（几百万行）

我提交工作来运行它在纱线集群（100个执行器）上，速度很慢，当我查看Spark UI中的DAG可视化时，似乎只有蜂巢表扫描任务并行运行，其余步骤＃2和＃3仅运行在一个哪些实例可能应该能够优化并行化？

该应用程序看起来像：

步骤1：

val input = hiveContext 
    .sql(
    SELECT 
      user_id 
      , address 
      , age 
      , phone_number 
      , first_name 
      , last_name 
      , server_ts 
     FROM 
     (  
      SELECT 
       user_id 
       , address 
       , age 
       , phone_number 
       , first_name 
       , last_name 
       , server_ts 
       , row_number() over 
       (partition by user_id, address, phone_number, first_name, last_name order by user_id, address, phone_number, first_name, last_name, server_ts desc, age) AS rn 
      FROM 
      ( 
       SELECT 
        user_id 
        , address 
        , age 
        , phone_number 
        , first_name 
        , last_name 
        , server_ts 
       FROM 
        table 
       WHERE 
        phone_number <> '911' AND 
        server_date >= '2015-12-01' and server_date < '2016-01-01' AND 
        user_id IS NOT NULL AND 
        first_name IS NOT NULL AND 
        last_name IS NOT NULL AND 
        address IS NOT NULL AND 
        phone_number IS NOT NULL AND 
      ) all_rows 
     ) all_rows_with_row_number 
     WHERE rn = 1) 

val input_tbl = input.registerTempTable(input_tbl) 

步骤2：

val result = hiveContext.sql(
    SELECT state, 
     phone_number, 
     address, 
     COUNT(*) as hash_count, 
     MAX(server_ts) as latest_ts 
    FROM 
    (SELECT 
     udf_getState(address) as state 
     , user_id 
     , address 
     , age 
     , phone_number 
     , first_name 
     , last_name 
     , server_ts 
    FROM 
     input_tbl) input 
    WHERE state IS NOT NULL AND state != '' 
    GROUP BY state, phone_number, address) 

步骤3：

result.cache() 
result.map(x => ...).saveAsNewAPIHadoopDataset(conf) 

的DAG可视化看起来像：

正如您所看到的，阶段0中的“过滤器”，“项目”和“交换”只在一个实例中运行，stage1和stage2也是如此，所以如果问题是几个问题和道歉哑：

1. 在每个执行器进行数据混洗之后，Driver中是否会出现“Filter”，“Project”和“Exchange”？
2. 什么代码映射到“过滤器”，“项目”和“交换”？
3. 我怎么可以同时运行“过滤器”，“项目”和“交换”来优化性能？
4. 可以同时运行stage1和stage2吗？

Answer 1

你不能正确读取DAG图 - 每个步骤是使用单箱并不意味着它不使用多个任务（因此核心）来计算可视化的事实步。

通过钻入舞台视图中可以看到每个步骤使用了多少个任务，该视图显示该阶段的所有任务。

例如，这里有一个样本DAG可视化与你相似：

你可以看到每个阶段由步骤“单”栏所示。

但是，如果我们看看下面的表格中，我们可以看到任务的每级数量：

其中之一是只使用2个任务，但其他使用220，这意味着数据被分割成220个分区，并且分区被并行处理，给定足够的可用资源。

如果深入了解该阶段，则可以再次看到它已使用了220个任务和所有任务的详细信息。从磁盘

只有任务读取数据显示在图中具有这些“多点”，以帮助您了解多少文件是如何被读取。

SO--正如Rashid的回答所暗示的，检查每个阶段的任务数量。

Answer 2

这不是很明显，所以我会做以下事情以解决问题。

1. 计算每个步骤的执行时间。
2. 如果您的表格是文本格式，则第一步可能会很慢，如果数据以实木复合地格式存储在Hive中，那么spark通常会更好。
3. 查看您的表是否由where子句中使用的列分区。
4. 如果将数据保存到Hbase的速度很慢，那么您可能需要预先拆分hbase表，因为默认情况下数据存储在单个区域中。
5. 看看阶段选项卡中的火花UI，看看有多少任务都开始为每一个阶段，也找数据地方一级描述here

希望，你将能够零的问题。