如何优化火花sql并行运行

Question

我是spark新手，有一个简单的spark应用程序，使用spark SQL/hiveContext：

1. 从hive表中选择数据（10亿行）
2. 做一些过滤，聚合，包括row_number窗口函数来选择第一行，分组，计数（）和最大（）等。
3. 将结果写入HBase（数亿行）

我提交的作业运行它在纱线集群（100个执行者），它很慢，当我在火花UI中查看DAG可视化时，似乎只有蜂巢表扫描任务并行运行，其余的步骤#2和#3只运行在一个实例中，可能应该能够优化并行化？

应用程序如下所示：

第一步：

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)

第二步：

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)

第三步：

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

如您所见，阶段0中的“过滤器”、“项目”和“交换”仅在一个实例中运行，阶段1和阶段2也是如此，因此，如果问题是愚蠢的，请回答几个问题并道歉：

1. 过滤器，项目和交换是否发生在驱动程序中，从每个执行程序的数据洗牌后？
2. 什么代码映射到过滤器，项目和交换？
3. 如何并行运行过滤器、项目和交换来优化性能？
4. 是否有可能并行运行第一阶段和第二阶段？

Answer 1

这是不明显的，所以我会做以下事情，以零的问题。

1. 计算每个步骤的执行时间

希望你能把问题归零。

Answer 2

您没有正确读取DAG图——每个步骤都是使用单个框可视化的事实并不意味着它没有使用多个任务（因此也就是核心）来计算该步骤。

您可以通过向下钻取阶段视图来查看每个步骤使用了多少任务，该阶段视图显示了该阶段的所有任务。

例如，这里有一个与您类似的DAG可视化示例：

您可以看到每个阶段都由“单个”步骤列描绘。

但如果我们看下表，我们可以看到每个阶段的任务数量：

其中一个只使用2个任务，而另一个使用220，这意味着数据被分割成220个分区，并且在提供足够可用资源的情况下并行处理分区。

如果你深入到这个阶段，你会再次看到它使用了220个任务和所有任务的细节。

只有从磁盘读取数据的任务才会在图中显示为具有这些“多个点”，以帮助您了解读取了多少文件。

正如拉希德的回答所建议的，检查每个阶段的任务数量。