--- title: 执行Spark引擎业务 description: "Spark使用交互式页面命令行来执行SQL任务，需要注意的是Spark侧看SparkExtension是否生效需要在SQL语句前加EXPLAIN语句，或者在Spark UI页面查看，如果算子是以Omni开头的则代表SparkExtension生效。" url: https://www.hikunpeng.com/document/detail/zh/kunpengboostkithistory/230RC2/bds/kunpengbds_omniruntime_20_0066.html sourcePath: /source/zh/kunpengboostkithistory/230RC2/bds/kunpengbds_omniruntime_20_0066.html indexId: b908d7ba4e858bdea805e23f2e54bda1995c765e53361348b3bbc0834e7a26c380 --- # 执行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启动命令如下。 ``` /usr/local/spark/bin/spark-sql --deploy-mode client --driver-cores 8 --driver-memory 40g --num-executors 30 --executor-cores 6 --executor-memory 35g --master yarn --conf spark.task.cpus=1 --conf spark.default.parallelism=600 --conf spark.sql.broadcastTimeout=500 --conf spark.sql.shuffle.partitions=600 --conf spark.sql.adaptive.enabled=true --database tpcds_bin_partitioned_orc_3 ``` SparkExtension插件启动命令如下。 ``` /usr/local/spark/bin/spark-sql --deploy-mode client --driver-cores 8 --driver-memory 40g --num-executors 24 --executor-cores 8 --executor-memory 5g --master yarn --conf spark.memory.offHeap.enabled=true --conf spark.memory.offHeap.size=35g --conf spark.task.cpus=1 --conf 'spark.driver.extraClassPath=/opt/omni-operator/lib/boostkit-omniop-spark-3.1.1-1.2.0-aarch64.jar:/opt/omni-operator/lib/boostkit-omniop-bindings-1.2.0-aarch64.jar:/opt/omni-operator/lib/dependencies/*' --conf 'spark.executor.extraClassPath=/opt/omni-operator/lib/boostkit-omniop-spark-3.1.1-1.2.0-aarch64.jar:/opt/omni-operator/lib/boostkit-omniop-bindings-1.2.0-aarch64.jar:/opt/omni-operator/lib/dependencies/*' --driver-java-options -Djava.library.path=/opt/omni-operator/lib --conf spark.sql.codegen.wholeStage=false --conf spark.executorEnv.LD_LIBRARY_PATH=/opt/omni-operator/lib:/usr/local/lib/HMPP --conf spark.executorEnv.OMNI_HOME=/opt/omni-operator --conf spark.driverEnv.LD_LIBRARY_PATH=/opt/omni-operator/lib:/usr/local/lib/HMPP --conf spark.driverEnv.OMNI_HOME=/opt/omni-operator --conf spark.executor.extraLibraryPath=/opt/omni-operator/lib --conf spark.driverEnv.LD_PRELOAD=/opt/omni-operator/lib/libjemalloc.so.2 --conf spark.executorEnv.LD_PRELOAD=/opt/omni-operator/lib/libjemalloc.so.2 --conf spark.sql.extensions=com.huawei.boostkit.spark.ColumnarPlugin --jars /opt/omni-operator/lib/boostkit-omniop-spark-3.1.1-1.2.0-aarch64.jar --jars /opt/omni-operator/lib/boostkit-omniop-bindings-1.2.0-aarch64.jar --conf spark.sql.orc.impl=native --conf spark.shuffle.manager=org.apache.spark.shuffle.sort.OmniColumnarShuffleManager --conf spark.omni.sql.columnar.fusion=false --conf spark.omni.sql.columnar.sortSpill.enabled=true --conf spark.omni.sql.columnar.sortSpill.rowThreshold=4000000 --conf spark.omni.sql.columnar.sortSpill.dirDiskReserveSize=214748364800 --conf spark.locality.wait=8 --conf spark.sql.autoBroadcastJoinThreshold=100M --conf spark.sql.broadcastTimeout=500 --conf spark.sql.cbo.enabled=false --conf spark.default.parallelism=600 --conf spark.sql.shuffle.partitions=600 --conf spark.sql.adaptive.enabled=true --conf spark.executorEnv.MALLOC_CONF=narenas:2 --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，若启用则需添加配置项\-\-conf spark.shuffle.manager="org.apache.spark.shuffle.sort.OmniColumnarShuffleManager"。ock启用请配置ock自有的shuffleManager类。默认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表示关闭。 | | 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.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 | 200000 | sort算子溢写触发条件，处理数据行超过此值触发溢写，单位为Byte。请根据实际环境的内存调整参数，可以适当增大此参数，从而减少sort算子溢写到磁盘文件的次数，减少磁盘IO操作。 | | spark.omni.sql.columnar.sortSpill.dirDiskReserveSize | 10737418240 | sort溢写磁盘预留可用空间大小，如果实际小于此值会抛异常，单位为Byte。根据实际环境的磁盘容量和业务场景调整参数，建议不超过业务数据大小，取值上限为实际环境的磁盘容量大小。 | | spark.omni.sql.columnar.sortSpill.enabled | false | sort算子是否开启溢写能力。true表示开启溢写能力，false表示关闭。 | | 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库。 | 2. 查看SparkExtension是否生效。 在SparkExtension和原生Spark-SQL交互式命令行窗口分别运行以下SQL语句。 建议同时启动两个命令行界面，并在每个窗口中分别启动SparkExtension和开源版本Spark-SQL，方便对比。 ``` 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语句。 ``` 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输出数据一致。