GaussDB(DWS)实践系列-ClickHouse数据迁移

#! /bin/bash/env python3

from pyspark.sql import SparkSession
from datetime import datetime, timedelta
import sys

def ch2dws():

    startTime = datetime.now()
    print("===== {}任务开始时间{}:".format("order2dws", startTime.strftime("%Y-%m-%d %H:%M:%S.%f")))

    spark = SparkSession.builder.master("yarn").appName("ch2dws").getOrCreate()

    sql = """
    select 
    kid
    , ym
    , ymareaid
    , amount
    , order_total
    from xxxxx.xxxxxx oia 
    where orderday='2021-06-03'
    limit 3000
    """
    print(sql)
    df = spark.read.format("jdbc").options(
        url="jdbc:clickhouse://xxxxx:8123/xxxxxxxx",
        driver="ru.yandex.clickhouse.ClickHouseDriver",
        dbtable="(%s) tp" % sql,
        user="xxxxxxxxx", password="xxxxxxxx").load()
    #  append
    df.write.mode(saveMode="overwrite") \
        .format("jdbc") \
        .option("driver", "org.postgresql.Driver") \
        .option("url", "jdbc:postgresql://xxxxxxxxx:8000/xxxxxxxxxx") \
        .option("useSSL", "false") \
        .option("user", "xxxxxxxxxx") \
        .option("password", "xxxxxxxxx") \
        .option("batchsize", 10000) \
        .option("dbtable", "xxxxxxxxxx") \
        .option("isolationLevel", "NONE") \
        .option("truncate", False) \
        .save()

    spark.stop()

    endTime = datetime.now()
    print("任务结束时间:{},总耗时{}".format(endTime, timedelta(seconds=(endTime - startTime).total_seconds())))

if __name__ == '__main__':
    # date_input = [it for it in sys.argv[1:]]
    ch2dws()