Spark算子下推使用spark-sql命令来执行。

本次任务示例使用tpch的1T数据的非分区表作为测试表，测试sql为tpch-sql6。

表1 相关表信息
表名	表格式	总数	占用空间
lineitem	orc	5999989709	169.6G

Spark 3.0.0具体步骤

运行spark-sql命令。

        
             spark-sql --num-executors 10 --executor-cores 6 --driver-class-path "/usr/local/spark-plugin-jar/*" --jars "/usr/local/spark-plugin-jar/*" --conf "spark.executor.extraClassPath=./*"

点击放大

执行sql6语句。

select
sum(l_extendedprice * l_discount) as revenue
from
tpch_flat_orc_1000.lineitem
where
l_shipdate >= '1993-01-01'
and l_shipdate < '1994-01-01'
and l_discount between 0.06 - 0.01 and 0.06 + 0.01
and l_quantity < 25;

执行任务时会打印下推的信息，如下所示。

ndp.NdpPushDown: Selectivity: 0.09795918367346941
ndp.NdpPushDown: Push down with [PushDownInfo(ListBuffer(FilterExeInfo((((((((isnotnull(l_quantity#5) AND isnotnull(l_discount#7)) AND isnotnull(l_shipdate#11)) AND (cast(l_shipdate#11 as date) >= 8401)) AND (cast(l_shipdate#11 as date) < 8766)) AND (l_discount#7 >= 0.05)) AND (l_discount#7 <= 0.07)) AND (l_quantity#5 < 25.0)),List(l_quantity#5, l_extendedprice#6, l_discount#7, l_shipdate#11))),ListBuffer(AggExeInfo(List(sum((l_extendedprice#6 * l_discount#7))),List(),List(sum#36))),None,Map(ceph1 -> ceph1, ceph2 -> ceph2, ceph3 -> ceph3))]

包含了下推的选择率以及算子信息。

点击放大

spark-sql命令参数信息如表2所示。

表2 算子下推的参数含义
参数	推荐值	含义
--num-executors	10	启动的执行器数量，缺省值为2。
--executor-cores	6	每个执行器使用的CPU核数，缺省值为1。
--driver-class-path	"/usr/local/spark-plugin-jar/*"	传递给驱动程序的额外JAR包的路径。
--jars	"/usr/local/spark-plugin-jar/*"	驱动程序和执行器类路径中要包含的JAR包。
--conf	"spark.executor.extraClassPath=./*"	配置Spark参数。

Spark 3.1.1步骤

由于Spark 3.1.1 yarn模式下不打印INFO级别的日志信息，所以Spark 3.1.1需要做日志重定向。

定义日志文件log4j.properties。

log4j.rootCategory=INFO, FILE
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.target=System.err
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{1}: %m%n
 
log4j.logger.org.apache.spark.sql.execution=DEBUG
log4j.logger.org.apache.spark.repl.Main=INFO
 
log4j.appender.FILE=org.apache.log4j.FileAppender
log4j.appender.FILE.file=/logs/file.log
log4j.appender.FILE.layout=org.apache.log4j.PatternLayout
 
log4j.appender.FILE.layout.ConversionPattern=%m%n

修改log4j.properties中的log4j.appender.FILE.file为自定义的目录和文件名。

运行spark-sql命令时加日志项--driver-java-options -Dlog4j.configuration=file:../conf/log4j.properties，例如。

/usr/local/spark/bin/spark-sql  --driver-class-path '/usr/local/spark-plugin-jar/*' --jars '/usr/local/spark-plugin-jar/*' --conf 'spark.executor.extraClassPath=./*' --name tpch_query6.sql --driver-memory 50G --driver-java-options -Dlog4j.configuration=file:../conf/log4j.properties --executor-memory 32G --num-executors 30 --executor-cores 18

执行Spark引擎业务

Spark 3.0.0具体步骤

Spark 3.1.1步骤