执行Spark引擎业务

Spark使用交互式页面命令行来执行SQL任务，需要注意的是Spark侧看SparkExtension是否生效需要在SQL语句前加EXPLAIN语句，或者在Spark UI页面查看，如果算子是以Omni开头的则代表SparkExtension生效。

本次任务示例使用tpcds_bin_partitioned_orc_3的数据表作为测试表，测试SQL为TPC-DS测试集的Q82。

相关的表信息如表1所示。

表1 相关表信息

表名

表格式

总行数

item

orc

36000

inventory

orc

28188000

date_dim

orc

73049

store_sales

orc

8639377
  1. 启动Spark-SQL命令行窗口。

    原生Spark-SQL启动命令如下。

    1
    		 
    /usr/local/spark/bin/spark-sql --deploy-mode client --driver-cores 8 --driver-memory 20g --master yarn --executor-cores 8 --executor-memory 26g --num-executors 36 --conf spark.executor.extraJavaOptions='-XX:+UseG1GC -XX:+UseNUMA' --conf spark.locality.wait=0 --conf spark.network.timeout=600 --conf spark.serializer=org.apache.spark.serializer.KryoSerializer --conf spark.sql.adaptive.enabled=true --conf spark.sql.autoBroadcastJoinThreshold=100M --conf spark.sql.broadcastTimeout=600 --conf spark.sql.shuffle.partitions=1000 --conf spark.sql.orc.impl=native --conf spark.task.cpus=1 --database tpcds_bin_partitioned_orc_3
    SparkExtension插件启动命令如下。 
    1
    		 
    /usr/local/spark/bin/spark-sql --deploy-mode client --driver-cores 8 --driver-memory 40g --conf spark.driverEnv.LD_LIBRARY_PATH=/opt/omni-operator/lib:/usr/local/lib/HMPP --conf spark.driverEnv.LD_PRELOAD=/opt/omni-operator/lib/libjemalloc.so.2 --conf spark.driverEnv.OMNI_HOME=/opt/omni-operator --conf spark.driver.extraClassPath=/opt/omni-operator/lib/boostkit-omniop-spark-3.1.1-1.3.0-aarch64.jar:/opt/omni-operator/lib/boostkit-omniop-bindings-1.3.0-aarch64.jar:/opt/omni-operator/lib/dependencies/* --conf spark.driver.extraLibraryPath=/opt/omni-operator/lib --driver-java-options -Djava.library.path=/opt/omni-operator/lib --master yarn --executor-cores 12 --executor-memory 5g --conf spark.memory.offHeap.enabled=true --conf spark.memory.offHeap.size=35g --num-executors 24 --conf spark.executorEnv.LD_LIBRARY_PATH=/opt/omni-operator/lib:/usr/local/lib/HMPP --conf spark.executorEnv.LD_PRELOAD=/opt/omni-operator/lib/libjemalloc.so.2 --conf spark.executorEnv.MALLOC_CONF=narenas:2 --conf spark.executorEnv.OMNI_HOME=/opt/omni-operator --conf spark.executor.extraClassPath=/opt/omni-operator/lib/boostkit-omniop-spark-3.1.1-1.3.0-aarch64.jar:/opt/omni-operator/lib/boostkit-omniop-bindings-1.3.0-aarch64.jar:/opt/omni-operator/lib/dependencies/* --conf spark.executor.extraJavaOptions='-XX:+UseG1GC -XX:+UseNUMA' --conf spark.executor.extraLibraryPath=/opt/omni-operator/lib --jars /opt/omni-operator/lib/boostkit-omniop-spark-3.1.1-1.3.0-aarch64.jar --jars /opt/omni-operator/lib/boostkit-omniop-bindings-1.3.0-aarch64.jar --conf spark.omni.sql.columnar.fusion=false --conf spark.serializer=org.apache.spark.serializer.KryoSerializer --conf spark.shuffle.manager=org.apache.spark.shuffle.sort.OmniColumnarShuffleManager --conf spark.sql.adaptive.enabled=true --conf spark.sql.autoBroadcastJoinThreshold=100M --conf spark.sql.broadcastTimeout=500 --conf spark.network.timeout=600 --conf spark.sql.codegen.wholeStage=false --conf spark.sql.extensions=com.huawei.boostkit.spark.ColumnarPlugin --conf spark.sql.orc.impl=native --conf spark.sql.shuffle.partitions=600 --conf spark.task.cpus=1 --database tpcds_bin_partitioned_orc_3

    上述启动命令为Yarn模式使用，若使用local模式启动SparkExtension插件，需将--master yarn改为--master local，同时在启动前需在所有节点的“~/.bashrc”文件中添加export LD_PRELOAD=/opt/omni-operator/lib/libjemalloc.so.2并更新环境变量，可参考6。

    SparkExtension相关的启动参数信息如表2所示。

    表2 SparkExtension相关启动参数信息

    启动参数名称

    缺省值

    含义

    spark.sql.extensions

    com.huawei.boostkit.spark.ColumnarPlugin

    启用SparkExtension。

    spark.shuffle.manager

    sort

    是否启用列式Shuffle，若启用请配置OmniShuffle自有的shuffleManager类，需添加配置项--conf spark.shuffle.manager="org.apache.spark.shuffle.sort.OmniColumnarShuffleManager"。默认sort走原生的Shuffle。

    spark.omni.sql.columnar.hashagg

    true

    是否启用列式HashAgg，true表示启用，false表示关闭。

    spark.omni.sql.columnar.project

    true

    是否启用列式Project，true表示启用，false表示关闭。

    spark.omni.sql.columnar.projfilter

    true

    是否启用列式ConditionProject（Project + Filter融合算子），true表示启用，false表示关闭。

    spark.omni.sql.columnar.filter

    true

    是否启用列式Filter，true表示启用，false表示关闭。

    spark.omni.sql.columnar.sort

    true

    是否启用列式Sort，true表示启用，false表示关闭。

    spark.omni.sql.columnar.window

    true

    是否启用列式Window，true表示启用，false表示关闭。

    spark.omni.sql.columnar.broadcastJoin

    true

    是否启用列式BroadcastHashJoin，true表示启用，false表示关闭。

    spark.omni.sql.columnar.nativefilescan

    true

    是否启用列式NativeFilescan，true表示启用，false表示关闭，包括ORC和Parquet的文件格式。

    spark.omni.sql.columnar.sortMergeJoin

    true

    是否启用列式SortMergeJoin，true表示启用，false表示关闭。

    spark.omni.sql.columnar.takeOrderedAndProject

    true

    是否启用列式TakeOrderedAndProject，true表示启用，false表示关闭。

    spark.omni.sql.columnar.shuffledHashJoin

    true

    是否启用列式ShuffledHashJoin，true表示启用，false表示关闭。

    spark.shuffle.columnar.shuffleSpillBatchRowNum

    10000

    Shuffle输出的每个batch中包含数据的行数。请根据实际环境的内存调整参数，可以适当增大此参数，从而减少写入磁盘文件的批次，提升写入速度。

    spark.shuffle.columnar.shuffleSpillMemoryThreshold

    2147483648

    Shuffle内存溢写上限，Shuffle内存上限达到缺省值时会发生溢写，单位：Byte。请根据实际环境的内存调整参数，可以适当增大此参数，从而减少Shuffle内存溢写到磁盘文件次数，减少磁盘IO操作。

    spark.omni.sql.columnar.sortMergeJoin.fusion

    false

    是否开启sortMergeJoin融合，true表示和sort子节点融合，false表示和sort子节点不融合。

    spark.shuffle.columnar.compressBlockSize

    65536

    Shuffle数据压缩块大小，单位：Byte。请根据实际环境的内存调整参数，建议采用缺省值。

    spark.sql.execution.columnar.maxRecordsPerBatch

    4096

    列式Shuffle初始化Buffer大小，单位：Byte。请根据实际环境的内存调整参数，可以适当增大此参数，从而减少Shuffle读写次数，提升性能。

    spark.shuffle.compress

    true

    Shuffle是否开启压缩。true表示压缩，false表示不压缩。

    spark.io.compression.codec

    lz4

    Shuffle压缩格式。支持uncompressed、zlib、snappy、lz4和zstd格式。

    spark.omni.sql.columnar.sortSpill.rowThreshold

    214783647

    sort算子溢写触发条件，处理数据行超过此值触发溢写，单位：行。请根据实际环境的内存调整参数，可以适当增大此参数，从而减少sort算子溢写到磁盘文件的次数，减少磁盘IO操作。

    spark.omni.sql.columnar.sortSpill.memFraction

    90

    sort算子溢写触发条件，处理数据使用堆外内存超过此百分比触发溢写，与堆外内存总大小参数“spark.memory.offHeap.size”同时使用。请根据实际环境的内存调整参数，可以适当增大此参数，从而减少sort算子溢写到磁盘文件的次数，减少磁盘IO操作。

    spark.omni.sql.columnar.broadcastJoin.shareHashtable

    true

    在Broadcast Join场景下，是否开启builder侧只构建一份hash table，并允许所有lookup join侧共用。true表示开启，false表示关闭。

    spark.omni.sql.columnar.sortSpill.dirDiskReserveSize

    10737418240

    sort溢写磁盘预留可用空间大小，如果实际小于此值会抛异常，单位：Byte。根据实际环境的磁盘容量和业务场景调整参数，建议不超过业务数据大小，取值上限为实际环境的磁盘容量大小。

    spark.omni.sql.columnar.sortSpill.enabled

    false

    sort算子是否开启溢写能力。true表示开启溢写能力，false表示关闭。

    spark.omni.sql.columnar.heuristicJoinReorder

    true

    是否开启join重排序优化策略。默认为true表示开启，false表示关闭。启发式join可以根据where过滤条件的数量和table表的大小，自动优化join的顺序。

    spark.default.parallelism=600

    200

    Spark并行执行的任务数。

    spark.sql.shuffle.partitions

    200

    Spark执行聚合操作或者Join操作时的Shuffle分区数。

    spark.sql.adaptive.enabled

    false

    是否启用自适应查询执行优化，可以在查询执行过程中动态地调整执行计划，true开启，false关闭。

    spark.executorEnv.MALLOC_CONF

    narenas:1

    控制Spark中每一个Executor进程中的内存分配策略。

    spark.sql.autoBroadcastJoinThreshold

    10M

    控制在执行Join操作时使用boradcastjoin小表的阈值大小。

    spark.sql.broadcastTimeout

    300

    控制广播小表到其它节点的超时时间。

    spark.omni.sql.columnar.fusion

    false

    是否把多个算子融合成一个算子。

    spark.locality.wait

    3

    数据本地化等待时长。

    spark.sql.cbo.enabled

    false

    是否开启CBO。

    spark.sql.codegen.wholeStage

    true

    是否开启全阶段代码生成。

    spark.sql.orc.impl

    native

    native表示使用原生版本的ORC库，hive表示使用Hive中的ORC库。

    spark.serializer

    空

    使用Kryo序列化。

    spark.executor.extraJavaOptions

    空

    Executor使用Hadoop本地库加速路径。

    spark.driver.extraJavaOptions

    空

    Driver使用Hadoop本地库加速路径。

    spark.network.timeout

    120

    所有网络交互的默认超时时间，单位：s。

  2. 查看SparkExtension是否生效。

    在SparkExtension和原生Spark-SQL交互式命令行窗口分别运行以下SQL语句。

     1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
    		 
    set spark.sql.adaptive.enabled=false;
explain select  i_item_id
    ,i_item_desc
    ,i_current_price
from item, inventory, date_dim, store_sales
where i_current_price between 76 and 76+30
and inv_item_sk = i_item_sk
and d_date_sk=inv_date_sk
and d_date between cast('1998-06-29' as date) and cast('1998-08-29' as date)
and i_manufact_id in (512,409,677,16)
and inv_quantity_on_hand between 100 and 500
and ss_item_sk = i_item_sk
group by i_item_id,i_item_desc,i_current_price
order by i_item_id
limit 100;

    SparkExtension输出执行计划如下图，如果算子以Omni开头则证明SparkExtension生效。

    原生Spark-SQL输出执行计划如下图。

  3. 运行SQL语句。

    在SparkExtension和原生Spark-SQL交互式命令行窗口分别运行以下SQL语句。

     1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
    		 
    select  i_item_id
    ,i_item_desc
    ,i_current_price
from item, inventory, date_dim, store_sales
where i_current_price between 76 and 76+30
and inv_item_sk = i_item_sk
and d_date_sk=inv_date_sk
and d_date between cast('1998-06-29' as date) and cast('1998-08-29' as date)
and i_manufact_id in (512,409,677,16)
and inv_quantity_on_hand between 100 and 500
and ss_item_sk = i_item_sk
group by i_item_id,i_item_desc,i_current_price
order by i_item_id
limit 100;

  4. 输出结果对比。

    • 原生Spark-SQL执行结果。

      执行计划如下。

    • SparkExtension插件执行结果。

      执行计划如下。

    • 结果对比。

      原生和SparkExtension的SQL输出数据一致。

父主题： 在Spark引擎上的应用