--- title: OmniShuffle Shuffle加速叠加OmniOperator算子加速特性在Spark引擎上的应用 description: "使用叠加特性之前请参见安装OmniOperator算子加速特性、安装OmniShuffle Shuffle加速特性完成OmniOperator和OmniShuffle安装。" url: https://www.hikunpeng.com/document/detail/zh/kunpengboostkithistory/230RC5/bds/kunpengbds_omniruntime_20_0145.html sourcePath: /source/zh/kunpengboostkithistory/230RC5/bds/kunpengbds_omniruntime_20_0145.html indexId: 93d9b03f2a2df9a3cf5b18a50b23cc8fd282395f6f9e03a1c1c0b3ddcfd47d4380 --- # OmniShuffle Shuffle加速叠加OmniOperator算子加速特性在Spark引擎上的应用 #### 前提条件 使用叠加特性之前请参见安装OmniOperator算子加速特性、安装OmniShuffle Shuffle加速特性完成OmniOperator和OmniShuffle安装。 #### 安装OmniShuffle Shuffle加速叠加OmniOperator算子加速特性 1. 从表3下载得到BoostKit-omnishuffle-spark-3.1.1-1.3.0-aarch64.zip，并上传至服务器的“/home/”目录下。 “/home”目录用户可自行定义。 2. 解压BoostKit-omnishuffle-spark-3.1.1-1.3.0-aarch64.zip，获取ock-omniop-shuffle-manager-23.0.0-for-spark-3.1.jar和ock-omniop-tuning-23.0.0-for-spark-3.1.jar，将两个JAR包拷贝到“${OCK_HOME}/jars”目录下，修改权限为550，并分发到所有子节点同样的目录下。 ``` unzip BoostKit-omnishuffle-spark-3.1.1-1.3.0-aarch64.zip mv ock-omniop-shuffle-manager-23.0.0-for-spark-3.1.jar $OCK_HOME/jars mv ock-omniop-tuning-23.0.0-for-spark-3.1.jar $OCK_HOME/jars chmod 550 $OCK_HOME/jars/ock-omniop-shuffle-manager-23.0.0-for-spark-3.1.jar $OCK_HOME/jars/ock-omniop-tuning-23.0.0-for-spark-3.1.jar scp ock-omniop-shuffle-manager-23.0.0-for-spark-3.1.jar ock-omniop-tuning-23.0.0-for-spark-3.1.jar xx.xx.xx.xx :$OCK_HOME/jars ``` xx.xx.xx.xx为各个子节点的IP地址。 #### 执行Spark引擎 1. 在/home目录下新建shuffle_operator.conf文件，关于shuffle_operator.conf文件中的参数说明请参考spark.conf。 a. 新建文件。 ``` vim /home/shuffle_operator.conf ``` b. 按“i”进入编辑模式，在文件中添加如下内容。 ``` spark.master yarn spark.task.cpus 1 spark.shuffle.compress true spark.shuffle.spill.compress true spark.rdd.compress true spark.executor.extraClassPath /usr/local/hive/lib/mysql-connector-java-8.0.13.jar:usr/local/hive/lib:/opt/ock/jars/*:/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/*:. spark.driver.extraClassPath /usr/local/hive/lib/mysql-connector-java-8.0.13.jar:usr/local/hive/lib:/opt/ock/jars/*:/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/*:. spark.driver.extraJavaOptions -Djava.library.path=/opt/ock/ucache/23.0.0/linux-aarch64/lib/common:/opt/ock/ucache/23.0.0/linux-aarch64/lib/common/ucx:/opt/ock/ucache/23.0.0/linux-aarch64/lib/common/ucx/ucx:/opt/ock/ucache/23.0.0/linux-aarch64/lib/datakit:/opt/ock/ucache/23.0.0/linux-aarch64/lib/mf:/opt/ock/jars:/opt/omni-operator/lib spark.executor.extraJavaOptions -Djava.library.path=/opt/ock/ucache/23.0.0/linux-aarch64/lib/common:/opt/ock/ucache/23.0.0/linux-aarch64/lib/common/ucx:/opt/ock/ucache/23.0.0/linux-aarch64/lib/common/ucx/ucx:/opt/ock/ucache/23.0.0/linux-aarch64/lib/datakit:/opt/ock/ucache/23.0.0/linux-aarch64/lib/mf:/opt/ock/jars:/opt/omni-operator/lib spark.driver.extraLibraryPath /home/:/home/sasl/lib/:/opt/ock/security/kmc_tool/lib/:/opt/ock/ucache/23.0.0/linux-aarch64/lib/common:/opt/ock/ucache/23.0.0/linux-aarch64/lib/common/ucx:/opt/ock/ucache/23.0.0/linux-aarch64/lib/common/ucx/ucx:/opt/ock/ucache/23.0.0/linux-aarch64/lib/datakit:/opt/ock/ucache/23.0.0/linux-aarch64/lib/mf:/opt/ock/jars:/opt/omni-operator/lib:. spark.executor.extraLibraryPath /home/:/home/sasl/lib/:/opt/ock/security/kmc_tool/lib/:/opt/ock/ucache/23.0.0/linux-aarch64/lib/common:/opt/ock/ucache/23.0.0/linux-aarch64/lib/common/ucx:/opt/ock/ucache/23.0.0/linux-aarch64/lib/common/ucx/ucx:/opt/ock/ucache/23.0.0/linux-aarch64/lib/datakit:/opt/ock/ucache/23.0.0/linux-aarch64/lib/mf:/opt/ock/jars:/opt/omni-operator/lib:. spark.shuffle.manager org.apache.spark.shuffle.ock.OckColumnarShuffleManager spark.shuffle.ock.manager true spark.broadcast.ock.manager false spark.blacklist.enabled true spark.files.fetchFailure.unRegisterOutputOnHost true spark.shuffle.service.enabled false spark.sql.adaptive.enabled true spark.blacklist.application.fetchFailure.enabled true spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version 2 spark.driver.maxResultSize 2g spark.serializer org.apache.spark.serializer.KryoSerializer spark.executorEnv.OCK_HOME /opt/ock spark.executorEnv.UCX_USE_MT_MUTEX y spark.executorEnv.UCX_TLS tcp spark.sql.broadcastTimeout 3000 #omni spark.executorEnv.LD_LIBRARY_PATH /opt/omni-operator/lib:/usr/local/lib/HMPP spark.executorEnv.OMNI_HOME /opt/omni-operator spark.driverEnv.LD_LIBRARY_PATH /opt/omni-operator/lib:/usr/local/lib/HMPP spark.driverEnv.OMNI_HOME /opt/omni-operator spark.driverEnv.LD_PRELOAD /opt/omni-operator/lib/libjemalloc.so.2 spark.executorEnv.LD_PRELOAD /opt/omni-operator/lib/libjemalloc.so.2 spark.sql.orc.impl native spark.omni.sql.columnar.fusion false spark.locality.wait 8 spark.sql.autoBroadcastJoinThreshold 100M spark.executorEnv.MALLOC_CONF narenas:2 spark.sql.extensions=com.huawei.boostkit.spark.ColumnarPlugin,org.apache.spark.sql.execution.adaptive.ock.OmniOpBoostTuningExtension spark.sql.ock.autoConfig.enabled true spark.sql.ock.autoConfig.history true spark.sql.ock.autoConfig.globalRuntimePartition false spark.sql.ock.autoConfig.sample false spark.kryoserializer.buffer.max 512m ``` c. 按“Esc”键，输入:wq!，按“Enter”保存并退出编辑。 2. 启动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插件启动命令如下。 ``` spark-sql --deploy-mode client --driver-cores 8 \ --driver-memory 40G \ --num-executors 24 \ --executor-cores 12 \ --executor-memory 5g \ --master yarn \ --conf spark.memory.offHeap.enabled=true \ --conf spark.memory.offHeap.size=26g \ --conf spark.sql.codegen.wholeStage=false \ --jars /opt/ock/jars/* \ --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 \ --properties-file /home/shuffle_operator.conf \ --database tpcds_bin_partitioned_orc_3 ``` 3. 查看OmniShuffle Shuffle加速叠加OmniOperator算子加速是否生效。 在SparkExtension和原生Spark-SQL交互式命令行窗口分别运行以下SQL语句。 ``` 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; ``` 输出执行计划如下图，如果算子中Shuffle算子变成BoostTuningOmniColumnarShuffleExchange则生效。