一、原理介绍
随着 SSD 的广泛应用及网卡性能的显著提升，Spark 引擎的性能瓶颈更多由传统认知的 IO 转变为以 CPU 为主的计算资源。而围绕 JVM 的 CPU 优化方案（如 Codegen ）存在诸多约束，如字节码长度、参数个数等存在限制，开发者也很难在 JVM 上利用现代 CPU 的一些特性。
Meson 引擎在 Spark 物理计划进行转换，使用 C++ 实现的向量化加速库执行计算，并将执行完的数据以列式方式返回，提升内存及带宽利用效率，进而突破性能瓶颈，可以使 Spark 作业效率得到有效的提升。
二、使用限制
Meson 引擎目前存在使用场景限制，在限制场景时 Meson 引擎将会进行 Fallback，回退到原生 Spark 引擎进行执行，由于 Fallback 需要进行必要的数据转换，回退次数过多，可能会导致总体运行时间比原生 Spark 引擎更慢。
请你提前了解 Meson 引擎主要的使用限制：
·支持 Parquet 数据格式，ORC 支持暂不完善，其他数据格式暂不支持。
·暂不支持 ANSI 模式。
·暂不支持基于 RDD 的 Application。
·暂不支持 Structured Streaming。
·暂不支持基于 PySpark 的自定义 Python 代码。
·暂不支持 MEMORY_ONLY 的 CacheTable。
三、适用场景
以下支持能力基于 Spark 3.5.3及以上版本提供。
注意：Meson 未完全支持或不支持的存储格式、数据类型、算子和函数会 Fallback 至原生 Spark 引擎执行。
四、存储格式
Meson 引擎支持的数据存储格式：
·支持的数据格式：Parquet、ORC
·支持的表格式： Iceberg、Hive
五、数据类型
Meson 引擎支持的数据类型：
·Byte、Short、Int、Long
·Boolean
·String、Binary
·Decimal
·Float、Double
·Date、Timestamp
六、算子
| 类型 | 已支持算子 | 未支持算子 |
| Source | FileSourceScanExec、HiveTableScanExec、BatchScanExec、InMemoryTableScanExec | – |
| Sink | DataWritingCommandExec、InsertIntoHiveTable、 | – |
| Common | FilterExec、ProjectExec、SortExec、UnionExec | – |
| Aggregate | HashAggregateExec | SortAggregateExec、ObjectHashAggregateExec |
| Join | BroadcastHashJoinExec、ShuffledHashJoinExec、SortMergeJoinExec、BroadcastNestedLoopJoinExec、CartesianProductExec | – |
| Window | WindowExec | WindowGroupLimitExec |
| Exchange | ShuffleExchangeExec、ReusedExchangeExec、BroadcastExchangeExec、CoalesceExec | CustomShuffleReaderExec |
| Limit | GlobalLimitExec、LocalLimitExec、TakeOrderedAndProjectExec、CollectLimitExec | – |
| Subquery | SubqueryBroadcastExec | – |
| Other | ExpandExec、GenerateExec、CollectTailExec、RangeExec | RangeExec、SampleExec |
七、函数
| 类型 | 已支持函数 |
| Generator Functions | explode,explode_outer,inline,inline_outer,posexplode,posexplode_outer,stack |
| Window Functions | cume_dist,dense_rank,lag,lead,nth_value,ntile,percent_rank,rank,row_number |
| Aggregate Functions | any,any_value,approx_count_distinct,approx_percentile,array_agg,avg,bit_and,bit_or,bit_xor,bool_and,bool_or,collect_list,collect_set,corr,count,count_if,covar_pop,covar_samp,every,first,first_value,grouping,grouping_id,kurtosis,last,last_value,max,max_by,mean,median,min,min_by,percentile,percentile_approx,regr_avgx,regr_avgy,regr_count,regr_intercept,regr_r2,regr_slope,regr_sxx,regr_sxy,regr_syy,skewness,some,std,stddev,stddev_pop,stddev_samp,sum,try_avg,try_sum,var_pop,var_samp,variance |
| Array Functions | array,array_append,array_compact,array_contains,array_distinct,array_except,array_insert,array_intersect,array_join,array_max,array_min,array_position,array_prepend,array_remove,array_repeat,array_union,arrays_overlap,arrays_zip,flatten,get,shuffle,slice,sort_array |
| Bitwise Functions | &,^,bit_count,bit_get,getbit,shiftright,|,~ |
| Collection Functions | array_size,cardinality,concat,reverse,size |
| Conditional Functions | coalesce,if,ifnull,nanvl,nullif,nvl,nvl2,when |
| Conversion Functions | bigint,binary,boolean,cast,date,decimal,double,float,int,smallint,string,timestamp,tinyint |
| Date and Timestamp Functions | add_months,date_add,date_diff,date_format,date_from_unix_date,date_sub,date_trunc,dateadd,datediff,day,dayofmonth,dayofweek,dayofyear,extract,from_unixtime,from_utc_timestamp,hour,last_day,make_date,make_timestamp,make_ym_interval,minute,month,next_day,quarter,second,timestamp_micros,timestamp_millis,to_unix_timestamp,to_utc_timestamp,trunc,unix_date,unix_micros,unix_millis,unix_seconds,unix_timestamp,weekday,weekofyear,year |
| Hash Functions | crc32,hash,md5,sha,sha1,sha2,xxhash64 |
| JSON Functions | from_json,get_json_object,json_array_length,json_object_keys,json_tuple,schema_of_json,to_json |
| Lambda Functions | aggregate,array_sort,exists,filter,forall,map_filter,map_zip_with,reduce,transform,transform_keys,transform_values,zip_with |
| Map Functions | element_at,map,map_concat,map_contains_key,map_entries,map_keys,map_values,str_to_map,try_element_at |
| Mathematical Functions | %,*,+,-,/,abs,acos,acosh,asin,asinh,atan,atan2,atanh,bin,cbrt,ceil,ceiling,conv,cos,cosh,cot,csc,degrees,e,exp,expm1,factorial,floor,greatest,hex,hypot,least,log,log10,log1p,log2,mod,negative,pi,pmod,positive,pow,power,rand,random,rint,round,sec,shiftleft,sign,signum,sinh,sqrt,try_add,unhex,width_bucket |
| Misc Functions | assert_true,equal_null,spark_partition_id,uuid,version,|| |
| Predicate Functions | !,!=,<,<=,<=>,<>,=,==,>,>=,and,between,case,ilike,in,isnan,isnotnull,isnull,like,not,or,regexp,regexp_like |
| String Functions | ascii,base64,bit_length,btrim,char,char_length,character_length,chr,concat_ws,contains,endswith,find_in_set,format_number,format_string,initcap,instr,lcase,left,len,length,levenshtein,locate,lower,lpad,ltrim,luhn_check,mask,overlay,position,regexp_extract,regexp_extract_all,regexp_replace,repeat,replace,right,rpad,rtrim,soundex,split,split_part,startswith,substr,substring,substring_index,translate,trim,ucase,unbase64,upper |
| Struct Functions | named_struct,struct |
| URL Functions | url_decode,url_encode |
八、EMR-V3.7.0
创建 EMR- V3.7.0 版本集群，你可以通过 EMR 控制台 配置管理 功能，在 spark-defaults.conf 配置文件中新增以下配置使用该特性：
| 参数 | 说明 |
| spark.plugins | Spark 用到的插件，参数值设置为 org.apache.gluten.GlutenPlugin（如果已经配置了 spark.plugins，则可以将 org.apache.gluten.GlutenPlugin 加到其中，用逗号","隔开） |
| spark.memory.offHeap.enabled | 设置为 true，Meson 加速需要用到 JVM 的 off memory |
| spark.memory.offHeap.size | 设置 offHeap 内存的大小，根据实际情况设置，详见不同规格executor内存配置推荐。 |
| spark.shuffle.manager | Meson 使用的列式 shuffle manager， 参数值设置为：org.apache.spark.shuffle.sort.ColumnarShuffleManager |
不同规格 Executor 内存配置推荐：
| executor-cores | spark.executor.memory | spark.memory.offHeap.size |
| 2 | 2GB | 4GB |
| 4 | 3GB | 10GB |
| 8 | 6GB | 20GB |
九、EMR-V3.6.1（beta）
创建 EMR- V3.6.1 版本集群，你可以通过 EMR 控制台 配置管理 功能，在 spark-defaults.conf 配置文件中新增以下配置使用该特性：
| 参数 | 说明 |
| spark.plugins | Spark 用到的插件，参数值设置为 org.apache.gluten.GlutenPlugin（如果已经配置了 spark.plugins，则可以将 org.apache.gluten.GlutenPlugin 加到其中，用逗号","隔开） |
| spark.memory.offHeap.enabled | 设置为 true，Native 加速需要用到 JVM 的 off memory |
| spark.memory.offHeap.size | 设置 offHeap 内存的大小，根据实际情况设置，初始可设置为1G |
| spark.shuffle.manager | Meson 使用的列式 shuffle manager， 参数值设置为：org.apache.spark.shuffle.sort.ColumnarShuffleManager |
| spark.driver.extraClassPath | Spark 用到的 Gluten native jar，jar 包默认路径在 /usr/local/service/spark/gluten 下 |
| spark.executor.extraClassPath | Spark 用到的 Gluten native jar，jar 包默认路径在 /usr/local/service/spark/gluten 下 |
| spark.executorEnv.LIBHDFS3_CONF | 集成的 HDFS 集群配置文件的路径，默认在/usr/local/service/hadoop/etc/hadoop/hdfs-site.xml |