折腾一晚上的事情，明白了一个道理

Posted 2022-09-07 硅谷工具人

感悟：有时候很简单的笨办法，比那些高大上的技术要实用的多。

有一个数据同步，大约4亿条记录，没有分区。现在要按照天，小时分区写入到iceberg的分区表中。
源数据中本身就是很多几十k大小的非常多的小文件。

于是在读取时，总想着要shuffle，合并小文件,于是是这样的：

hive_df = spark.table("xx.hive_test")
hive_df.repartition(1).write.format('iceberg').mode('overwrite').save('iceberg_dw.iceberg_test')

但是iceberg表写入分区表时，需要根据分区本地排序，或者全局排序，于是写成这样：

hive_df = spark.table("xx.hive_test")
hive_df.repartition(1).sortWithinPartitions("pk_day","pk_hour").write.format('iceberg').mode('overwrite').save('iceberg_dw.iceberg_test')

但是执行起来后，分区排序合并都没有问题，但是由于数据量巨大，都卡在最后的1个partition任务那。于是又改成了多个重分区。

hive_df = spark.table("xx.hive_test")
hive_df.repartition(20).sortWithinPartitions("pk_day","pk_hour").write.format('iceberg').mode('overwrite').save('iceberg_dw.iceberg_test')

但是依然很慢，于是加大了分区

hive_df = spark.table("xx.hive_test")
hive_df.repartition(100).sortWithinPartitions("pk_day","pk_hour").write.format('iceberg').mode('overwrite').save('iceberg_dw.iceberg_test')

折腾了大半夜，通过增大excuter个数，内存大小还是不行。

早上醒来，忽然想到了大数据的分而治之的想法，以前跑数据的时候也干过，将源头数据分几段来读取写入，多跑几个任务就可以。
于是修改为以下代码，通过传入年月日，时分秒，再转换为对应的时间戳取查询源头的记录，将单个小时的合并为1个分区写入目标表。
pySpark程序命名为hive2iceberg.py

from pyspark.sql import SparkSession
import sys,datetime,time,pytz
day_hour = sys.argv[1] + ' ' + sys.argv[2] # 2022-09-02 09:00:00
utc_timestamp = time.mktime(datetime.datetime.strptime(day_hour, '%Y-%m-%d %H:%M:%S').replace(tzinfo=pytz.utc).timetuple())
dt = int(utc_timestamp) * 1000
pk_day = dt - dt % 86400000
dayBeforeYesterday = (dt - dt % 86400000) - 86400000
pk_hour = dt - dt % 3600000

print("=====================================================================================================================")
print(">>>>>>>>>>>>>>>>日期小时:" + str(day_hour) +  " pk_day:" + str(pk_day), "dayBeforeYesterday:" + str(dayBeforeYesterday) + " hour:" + str(pk_hour))
print("=====================================================================================================================")

pyfilename= "hive2iceberg"
spark = SparkSession \\
    .builder \\
    .config("spark.sql.shuffle.partitions", "1000") \\
    .config("spark.driver.maxResultSize", "6g") \\
    .config("spark.debug.maxToStringFields", "1000") \\
    .config("spark.sql.iceberg.handle-timestamp-without-timezone", "true") \\
    .config("spark.sql.session.timeZone", "UTC") \\
    .config("spark.sql.crossJoin.enabled","true") \\
    .config("spark.sql.catalog.iceberg", "org.apache.iceberg.spark.SparkCatalog") \\
    .config("iceberg.engine.hive.enabled","true") \\
    .config("hive.exec.dynamic.partition", "true") \\
    .config("hive.exec.dynamic.partition.mode", "nonstrict") \\
    .config("hive.exec.max.dynamic.partitions", "5000") \\
    .appName(pyfilename) \\
    .enableHiveSupport() \\
    .getOrCreate()

df = spark.table("xx.hive_test").where("pk_day = %d and pk_hour = %d " % (pk_day,pk_hour))
df.repartition(1).sortWithinPartitions("pk_day","pk_hour").write.format('iceberg').mode('overwrite').save('iceberg_dw.iceberg_test')

然后又准备了一段shell脚本run_state_hive2iceberg_fixdata.sh如下：

#!/bin/sh

if [ $# = 1 ]
then
    start_dt=$1
    end_dt=$1

elif [ $# = 2 ]
then
    start_dt=$1
    end_dt=$2

elif [ $# = 0 ]
then
    start_dt=`date -d '-1 hour' +%Y%m%d-%H`
    end_dt=`date -d '-1 hour' +%Y%m%d-%H`
fi

date_format_1=$start_dt:0:4-$start_dt:4:2-$start_dt:6:2' '$start_dt:9:2:00:00
date_format_2=$end_dt:0:4-$end_dt:4:2-$end_dt:6:2' '$end_dt:9:2:00:00

start_sec=`date -d "$date_format_1" +%s`
end_sec=`date -d "$date_format_2" +%s`


for ((i=$start_sec;i<=$end_sec;i+=3600))
do

file_day=`date -d @$i +%Y%m%d`
data_day=`date -d @$i +%Y-%m-%d`
data_hour=`date -d @$i +"%H:00:00"`

echo $data_day $data_hour

spark-submit \\
--conf spark.dynamicAllocation.enabled=false \\
--name hive2iceberg \\
--master yarn \\
--deploy-mode cluster \\
--queue prod \\
--driver-memory 2G \\
--num-executors 5 \\
--executor-memory 5G \\
--executor-cores 2 \\
--archives hdfs://ns1/user/hadoop/mypy3spark_env/py3spark.tar.gz#py3spark \\
--conf "spark.pyspark.python=./py3spark/py3spark/bin/python" \\
--conf "spark.pyspark.driver.python=./py3spark/py3spark/bin/python" \\
/home/hadoop/hive2iceberg.py $data_day $data_hour

status=$?

if [ $status = 0 ]; then
       echo "STATUS=SUCCESS"
else
       echo "STATUS=FAIL"
fi

查询源表里面的最大时间和最小时间，然后将时间切割成多个数据段，于是得到下面的执行脚本，因为上面的shell是循环按年月日小时跑的，所以放在后台执行即可。

nohup ./run_state_hive2iceberg_fixdata.sh 2021050504 2021070100 > 2021070100.log 2>&1 &
nohup ./run_state_hive2iceberg_fixdata.sh 2021070100 2021090100 > 2021090100.log 2>&1 &
nohup ./run_state_hive2iceberg_fixdata.sh 2021090100 2021110100 > 2021110100.log 2>&1 &
nohup ./run_state_hive2iceberg_fixdata.sh 2021110100 2022010100 > 2022010100.log 2>&1 &
nohup ./run_state_hive2iceberg_fixdata.sh 2022010100 2022040100 > 2022040100.log 2>&1 &
nohup ./run_state_hive2iceberg_fixdata.sh 2022040100 2022060100 > 2022060100.log 2>&1 &
nohup ./run_state_hive2iceberg_fixdata.sh 2022060100 2022080100 > 2022080100.log 2>&1 &
nohup ./run_state_hive2iceberg_fixdata.sh 2022080100 2022090220 > 2022090220.log 2>&1 &

自己大概计算了下，一个小时的数据量很小，从hive读取到写入iceberg也就1分钟左右，所以一个小时的时间可以跑数据的大概60个小时。也就是一天可以跑数据的1440个小时，也就是2个月。
那么从2021年5月-2022年9月，大概17个月，我分成了8个任务，每个任务跑2个月，大概全部跑完，也就一天的时间。
而且实际计算了下，使用的总算力比跑一个大任务还少不少。

现在想想有时候真的越笨的方法，越是简单轻松。
好了，任务挂到后台，可以补觉去了。