时间戳

首页 > 解决方案 > 火花 sql 执行 except() 查询需要很多时间

apache-spark - 火花 sql 执行 except() 查询需要很多时间

问题描述

我在火花应用程序中有以下代码

val log_file_df = loadInputFile() /*loads log file as Dataframe */
val split_df: DataFrame = splitLineByDelimiter(log_file_df) /* applies filter 
                                                    function to input file */
val bad_data_df: DataFrame = parseAndSaveBadData(split_df) /*filters split dataframe into bad data */

val good_data_df = split_df.except(bad_data_df) /* separates good data from bad data */

如果我在 split_df,bad_data_df 上运行诸如 show() 之类的任何操作，它的执行时间会更短（大约 1.5 分钟），并且我检查了物理计划输入日志文件是否只读一次

但是如果我对好的数据执行任何操作，它会花费更多的时间。（4 分钟）

val good_data_df = split_df.except(bad_data_df).show()

从物理计划输入日志文件读取两次。我尝试了以下选项

    split_df.cache() or split_df.createOrReplaceTempView("split_dfTable")
//   Init.getSparkSession.sqlContext.cacheTable("split_dfTable")
     val splitbc =  Init.getSparkSession.sparkContext.broadcast(split_df)

但是执行时间没有改善，物理计划是一样的。这是物理计划。我应该如何改进我的代码？我的 good_data_df 被进一步转换并与其他一些需要更多时间的数据框连接。

good_data_df.show(false)good_data_df.explain(true)

 +- Exchange hashpartitioning(hostname#16, date#17, path#18, status#19,
    content_size#20, 200)
 +- *HashAggregate(keys=[hostname#16, date#17, path#18, status#19, 
      content_size#20], functions=[], output=[hostname#16, date#17, path#18, 
       status#19, content_size#20])
   +- SortMergeJoin [coalesce(hostname#16, ), coalesce(date#17, ), 
                   coalesce(path#18, ), coalesce(status#19, ), 
    coalesce(content_size#20, )], 
[coalesce(hostname#49, ), coalesce(date#50, ), coalesce(path#51, ), 
coalesce(status#52, ), coalesce(content_size#53, )], LeftAnti, 
(((((hostname#16 <=> hostname#49) && (date#17 <=> date#50)) && (path#18 <=> 
path#51)) && (status#19 <=> status#52)) && (content_size#20 <=> 
content_size#53))
:- *Sort [coalesce(hostname#16, ) ASC NULLS FIRST, coalesce(date#17, ) ASC 
NULLS FIRST, coalesce(path#18, ) ASC NULLS FIRST, coalesce(status#19, ) ASC 
NULLS FIRST, coalesce(content_size#20, ) ASC NULLS FIRST], false, 0
:  +- Exchange hashpartitioning(coalesce(hostname#16, ), coalesce(date#17, ), 
coalesce(path#18, ), coalesce(status#19, ), coalesce(content_size#20, ), 200)
:     +- *Project [regexp_extract(val#13, ^([^\s]+\s), 1) AS hostname#16, 
regexp_extract(val#13, ^.*(\d\d/\w{3}/\d{4}:\d{2}:\d{2}:\d{2} -\d{4}), 1) AS 
date#17, regexp_extract(val#13, ^.*"\w+\s+([^\s]+)\s*[(HTTP)]*.*", 1) AS 
path#18, regexp_extract(val#13, ^.*"\s+([^\s]+), 1) AS status#19, 
regexp_extract(val#13, ^.*\s+(\d+)$, 1) AS content_size#20]
:        +- *FileScan csv [val#13] Batched: false, Format: CSV, Location: 
InMemoryFileIndex[file:/C:/Users/M1047320/Desktop/access_log_Jul95], 
PartitionFilters: [], PushedFilters: [], ReadSchema: struct<val:string>
+- *Sort [coalesce(hostname#49, ) ASC NULLS FIRST, coalesce(date#50, ) ASC 
NULLS FIRST, coalesce(path#51, ) ASC NULLS FIRST, coalesce(status#52, ) ASC 

NULLS FIRST, coalesce(content_size#53, ) ASC NULLS FIRST], false, 0
        +- Exchange hashpartitioning(coalesce(hostname#49, ), 
coalesce(date#50, ), coalesce(path#51, ), coalesce(status#52, ), 
coalesce(content_size#53, ), 200)
+- *Project [regexp_extract(val#13, ^([^\s]+\s), 1) AS hostname#49, 
regexp_extract(val#13, ^.*(\d\d/\w{3}/\d{4}:\d{2}:\d{2}:\d{2} -\d{4}), 1) AS 
date#50, regexp_extract(val#13, ^.*"\w+\s+([^\s]+)\s*[(HTTP)]*.*", 1) AS 
path#51, regexp_extract(val#13, ^.*"\s+([^\s]+), 1) AS status#52, 
regexp_extract(val#13, ^.*\s+(\d+)$, 1) AS content_size#53]
              +- *Filter ((((regexp_extract(val#13, ^.*"\w+\s+([^\s]+)\s* 
[(HTTP)]*.*", 1) RLIKE .*(jpg|gif|png|xbm|jpeg|wav|mpg|pl)$ || 
(regexp_extract(val#13, ^([^\s]+\s), 1) = )) || (regexp_extract(val#13, 
^.*"\w+\s+([^\s]+)\s*[(HTTP)]*.*", 1) = )) || (regexp_extract(val#13, ^.* 
(\d\d/\w{3}/\d{4}:\d{2}:\d{2}:\d{2} -\d{4}), 1) = )) || 
(regexp_extract(val#13, ^.*"\s+([^\s]+), 1) = ))
                 +- *FileScan csv [val#13] Batched: false, Format: CSV,
Location: 
InMemoryFileIndex[file:/C:/Users/M1047320/Desktop/access_log_Jul95], 
PartitionFilters: [], PushedFilters: [], ReadSchema: struct<val:string>

标签： apache-sparkapache-spark-sql

解决方案

显示 split_df,bad_data_df 花费的时间要少得多的原因是Spark 只会读取和解析 show 所需的行。您读入 Spark 的数据被分解为Partitions，这些数据片段将在工作人员之间划分。

一旦你在 split_df 上调用 show，bad_data_df Spark 将只对一小部分数据起作用（split_df 上只有 20 行，bad_data_df 上只有前 20 行坏行）。

另一方面，当在 good_data_df 上调用 show 时，Spark 将不得不处理所有数据（以读取所有数据、解析数据以及从总数中排除坏行为代价）。

如果您有一种定义坏行的简单方法，我建议添加另一列带有UDF - Boolean isBad，并对其进行过滤。简单地传递数据比except.

推荐阅读