20 - MLSQL on k8s（7） - 集成RSS

MLSQL之道 2021-09-24

661

RSS（Spark Remote Shuffle Service），社区很早就有探讨过，但是一直没有做。随着K8S越来越火，而且已经支持Spark on K8S，因此对RSS的需求更为迫切。目前Spark on K8S的Shuffle存在一些问题，笔者列了几个：

Shuffle的磁盘问题，本地磁盘还是网络存储
缺乏隔离（比如：因为Shuffle导致跑在Yarn上的其他App受到影响）
可扩张性差（比如：缺乏副本导致一个失败的Node重算Lineage）

详细的讨论可以查看以下链接：

https://docs.google.com/document/d/1uCkzGGVG17oGC6BJ75TpzLAZNorvrAU3FRd2X-rVHSM/edit?ts=5e3c57b8#

为了解决这些问题，各个大厂提出了自己的Spark Shuffle解决方案，笔者给大家罗列一下：

1 uber https://databricks.com/session_na20/zeus-ubers-highly-scalable-and-distributed-shuffle-as-a-service
2 linkedin https://engineering.linkedin.com/blog/2020/introducing-magnet
3 facebook https://bestoreo.github.io/post/cosco/cosco/
4 趣头条和阿里 https://www.sohu.com/a/436931403_612370
5 有赞 https://iteblog.blog.csdn.net/article/details/114267593
6 京东 https://m.sohu.com/a/447193430_315839

目前，Uber开源了RSS（linkedin也有开源计划），但是功能还不成熟（比如：Shuffle文件清理策略，而且笔者还测出了Shuffle的Bug导致磁盘持续暴涨），也不完善（比如：Shuffle不支持多磁盘，不支持HDFS等存储），而且不活跃，但是做个示例还是很不错的。如果要使用的话，需要二次开发。今天，笔者用Uber开源的RSS，来和MLSQL做个集成示例。

https://github.com/uber/RemoteShuffleService

笔者按照使用文档安装，在executor端总是报找不到类的错误，就是找不到Jar包，因此笔者在构建镜像的时候，把remote-shuffle-service-0.0.9-client.jar放到Spark的jars下，解决了这个问题。

Dockerfile如下：

cat > MLSQLDockerfile << EOF
FROM 172.16.2.66:5000/spark:3.0-j14-mlsql
USER root
RUN useradd -ms bin/sh  hdfs
COPY streamingpro-mlsql-spark_3.0_2.12-2.1.0-SNAPSHOT.jar opt/spark/work-dir/
COPY remote-shuffle-service-0.0.9-client.jar opt/spark/jars/
COPY addHost.sh opt/spark/work-dir/
RUN sed -i '20 r /opt/spark/work-dir/addHost.sh' /opt/entrypoint.sh
WORKDIR /opt/spark/work-dir
EOF


docker build -t 172.16.2.66:5000/mlsql:3.0-j14-mlsql -f MLSQLDockerfile .
docker push 172.16.2.66:5000/mlsql:3.0-j14-mlsql

启动RSS Server：

java -Dlog4j.configuration=log4j-rss-prod.properties -cp remote-shuffle-service-0.0.9-server.jar com.uber.rss.StreamServer -port 12222 -serviceRegistry standalone -dataCenter dc1 -appMemoryRetentionMillis 60000 -appFileRetentionMillis 3600000
#最后两个参数是设置Shuffle数据在内存保存的毫秒和文件保存的毫秒。具体参数可以查看类：StreamServerConfig

MLSQL启动脚本：

cat > mlsql-start-rss.sh << EOF 
ip=$(cat etc/hosts | head -n 8 | tail -n 1 | awk '{print $1}')
echo $ip


/opt/spark/bin/spark-submit --master k8s://https://172.16.2.62:6443 \
            --deploy-mode client     \
            --class streaming.core.StreamingApp  \
            --conf spark.kubernetes.container.image=172.16.2.66:5000/mlsql:3.0-j14-mlsql \
            --conf spark.kubernetes.container.image.pullPolicy=Always  \
            --conf spark.kubernetes.namespace=default                            \
            --conf spark.kubernetes.executor.request.cores=0.05  \
            --conf spark.kubernetes.executor.limit.cores=0.3                 \
            --conf spark.dynamicAllocation.enabled=true  \
            --conf spark.dynamicAllocation.shuffleTracking.enabled=true \
            --conf spark.dynamicAllocation.minExecutors=1 \
            --conf spark.dynamicAllocation.maxExecutors=2 \
            --conf spark.dynamicAllocation.executorIdleTimeout=60 \
            --conf spark.shuffle.manager=org.apache.spark.shuffle.RssShuffleManager \
            --conf spark.shuffle.rss.serviceRegistry.type=standalone \
            --conf spark.shuffle.rss.serviceRegistry.server=172.16.2.62:12222 \
            --conf spark.shuffle.rss.dataCenter=dc1 \
            --conf spark.jars.ivy=/tmp/.ivy \
            --conf spark.driver.host=$ip \
            --conf spark.sql.cbo.enabled=true  \
            --conf spark.sql.adaptive.enabled=true  \
            --conf spark.sql.cbo.joinReorder.enabled=true  \
            --conf spark.sql.cbo.planStats.enabled=true  \
            --conf spark.sql.cbo.starSchemaDetection=true  \
            --conf spark.driver.maxResultSize=512m  \
            --conf spark.executor.memory=512m \
            --conf spark.driver.memory=512m \
            --conf spark.serializer=org.apache.spark.serializer.KryoSerializer  \
            --conf spark.kryoserializer.buffer.max=100m \
            --conf spark.executor.extraJavaOptions="-XX:+UnlockExperimentalVMOptions -XX:+UseZGC -XX:+UseContainerSupport  -Dio.netty.tryReflectionSetAccessible=true" \
            --conf spark.driver.extraJavaOptions="-XX:+UnlockExperimentalVMOptions -XX:+UseZGC -XX:+UseContainerSupport  -Dio.netty.tryReflectionSetAccessible=true -DREALTIME_LOG_HOME=/tmp/__mlsql__/logs" \
            --jars opt/mlsql/jar/juicefs-hadoop-0.11.0.jar \
            opt/mlsql/jar/streamingpro-mlsql-spark_3.0_2.12-2.1.0-SNAPSHOT.jar \
            -streaming.name mlsql \
            -streaming.rest true  \
            -streaming.thrift false  \
            -streaming.platform spark  \
            -streaming.enableHiveSupport true  \
            -streaming.spark.service true  \
            -streaming.job.cancel true              \
            -streaming.driver.port 9003
EOF