场景如下：

我有一个SparkSQL程序，它对几个Hive表执行ETL过程。这些表是使用RAW TEXT中的Sqoop和Snappy压缩从Teradata数据库导入的（不幸的是Avro格式不适用于Teradata连接器）。SparkSQL过程完成所需的时间约为1小时15分钟。

为了提高性能，我想在执行SparkSQL过程之前，将表转换为更有效的格式，比如Parquet。根据留档和在线讨论，这应该会给使用原始文本带来显著的提升（即使使用snappy压缩，原始文本也不能拆分）。因此，我使用Snappy压缩将所有Hive表转换为Parquet格式。我已经在这些表上启动了相同设置的SparkSQL过程（num-执行器、驱动程序-内存、执行器-内存）。这个过程在1小时20分钟内结束。这对我来说非常令人惊讶。我没想到会像我在一些讨论中读到的那样提高30倍，但我当然期待改进。

在Spark程序中执行的操作类型大部分是连接和过滤器（其中有条件），如以下片段所示：

val sc = new SparkContext(conf)
val sqc = new HiveContext(sc)
sqc.sql("SET hive.exec.compress.output=true")
sqc.sql("SET parquet.compression=SNAPPY")


var vcliff = sqc.read.table(s"$swamp_db.DBU_vcliff")
var vtktdoc = sqc.read.table(s"$swamp_db.DBU_vtktdoc")
var vasccrmtkt = sqc.read.table(s"$swamp_db.DBU_vasccrmtkt")

val numPartitions = 7 * 16

// caching
vcliff.registerTempTable("vcliff")
vtktdoc.registerTempTable("vtktdoc")
vasccrmtkt.registerTempTable("vasccrmtkt")


ar ORI_TktVCRAgency = sqc.sql(
    s"""
       |            SELECT tic.CODCLI,
       |            tic.CODARLPFX,
       |            tic.CODTKTNUM,
       |            tic.DATDOCISS,
       |            vloc.CODTHR,
       |            vloc.NAMCMPNAMTHR,
       |            vloc.CODAGNCTY,
       |            vloc.NAMCIT,
       |            vloc.NAMCOU,
       |            vloc.CODCOU,
       |            vloc.CODTYPTHR,
       |            vloc.CODZIP,
       |            vcom.CODCOMORGLEVDPC,
       |            vcom.DESCOMORGLEVDPC,
       |            vcom.CODCOMORGLEVRMX,
       |            vcom.DESCOMORGLEVRMX,
       |            vcom.CODCOMORGLEVSALUNT,
       |            vcom.CODPSECOMORGCTYLEVSALUNT,
       |            vcom.DESCOMORGLEVSALUNT,
       |            vcom.CODCOMORGLEVRPR,
       |            vcom.CODPSECOMORGCTYLEVRPR,
       |            vcom.DESCOMORGLEVRPR,
       |            vcom.CODCOMORGLEVCTYCNL,
       |            vcom.CODPSECOMORGCTYLEVCTYCNL,
       |            vcom.DESCOMORGLEVCTYCNL,
       |            vcom.CODCOMORGLEVUNT,
       |            vcom.CODPSECOMORGCTYLEVUNT,
       |            vcom.DESCOMORGLEVUNT,
       |            vcli.DESCNL
       |            FROM $swamp_db.DBU_vlocpos vloc
       |                LEFT JOIN $swamp_db.DBU_vcomorghiemktgeo vcom ON vloc.codtypthr = vcom.codtypthr
       |            AND vloc.codthr = vcom.codthr
       |            LEFT JOIN TicketDocCrm tic ON tic.codvdt7 = vloc.codthr
       |            LEFT JOIN vcliff vc ON vc.codcli = tic.codcli
       |            LEFT JOIN $swamp_db.DBU_vclieml vcli ON vc.codcli = vcli.codcli
     """.stripMargin)

ORI_TktVCRAgency.registerTempTable("ORI_TktVCRAgency")

[...]

var TMP_workTemp = sqc.sql(
    s"""
       |SELECT *
       |FROM TicketDocCrm
       |            WHERE CODPNRREF != ''
       |            AND (DESRTGSTS LIKE '%USED%'
       |            OR DESRTGSTS LIKE '%OK%'
       |            OR DESRTGSTS LIKE '%CTRL%'
       |            OR DESRTGSTS LIKE '%RFND%'
       |            OR DESRTGSTS LIKE '%RPRT%'
       |            OR DESRTGSTS LIKE '%LFTD%'
       |            OR DESRTGSTS LIKE '%CKIN%')
     """.stripMargin)

TMP_workTemp.registerTempTable("TMP_workTemp")

var TMP_workTemp1 = sqc.sql(
    s"""
       |SELECT *
       |FROM TMP_workTemp w
       |INNER JOIN
       |    (SELECT CODTKTNUM as CODTKTNUM_T
       |    FROM (
       |        SELECT CODCLI, CODTKTNUM, COUNT(*) as n
       |        FROM TMP_workTemp
       |        GROUP BY CODCLI, CODTKTNUM
       |        HAVING n > 1)
       |    a) b
       |ON w.CODTKTNUM = b.CODTKTNUM_T
     """.stripMargin).drop("CODTKTNUM_T")

[...]

集群由2个master和7个worker组成。每个节点都有：

• 16核cpu
• 110 GB内存

火花在YARN。

有人知道为什么在Spark中处理数据之前，从原始文本切换到Parquet格式没有得到任何性能改进吗？