pyspark从hive导出数据到clickhouse(pyspark杩炴帴hive)

需求

从hive导出数据到clickhouse 大概有两百个字段，两千多万数据，30G左右

实现

采用pyspark导入，先上代码

#_*_ coding=UTF-8 _*_
import os
import sys
from pyspark.sql import SparkSession


def sparksession_read_hive():  # .master("local[*]")
    spark = SparkSession.builder.appName('bi_pharaoh_mkt_lead_detail_to_clickhouse')
        .enableHiveSupport().getOrCreate()
    hive_df = spark.sql("""select  
    place_type 
                ,sub_place_type 
                
    from  dm.test where pt = 20220314
    """)

    print(hive_df.show())
    print('sparkcontext_read_hive读取完毕，开始准备写入')
    properties = {'driver': 'ru.yandex.clickhouse.ClickHouseDriver',
                  "socket_timeout": "300000",
                  "rewriteBatchedStatements": "true",
                  "batchsize": "1000000",
                  "numPartitions": "8",
                  'user': 'analysis_superset',
                  'password': 'JEjGnIj96VVh9a0h',
                  'isolationLevel': 'NONE'
                  }

    hive_df.write.jdbc(url='jdbc:clickhouse://{url}:8123/analysis',
                       table='test_table', mode='append', properties=properties)
    print('sparksession_read_hive写入完成')
    spark.stop()


def spark_te():
    spark = SparkSession.builder.appName('clickhouse').getOrCreate()
    properties = {"driver": "ru.yandex.clickhouse.ClickHouseDriver",
                  "socket_timeout": "300000",
                  "rewriteBatchedStatements": "true",
                  "batchsize": "1000000",
                  "numPartitions": "8",
                  'user': 'analysis_superset',
                  'password': 'JEjGnIj96VVh9a0h'}
    df = spark.read.jdbc(url='jdbc:clickhouse://cc-2ze6h5d90y45bsizb.clickhouse.ads.aliyuncs.com:8123/analysis',
                         table='bi_pharaoh_mkt_lead_detail_da', properties=properties)
    # spark.read.jdbc(properties=properties)
    # print(spark.sql('select * from default.test'))
    print(df.show(10))


if __name__ == '__main__':
    sparksession_read_hive()
    # spark_te()

再详细解释下，上面sparksession_read_hive() 这个方法是写入的方法，spark_te()这个方法是读取clickhouse的方法。spark简单语法如下：
SparkSession是新版本推荐的入口api，所有的spark操作都需要SparkSession来执行，首先，创建一个接口，(类似于python的实例化)
语法是这样的：SparkSession.builder 然后，有下面几个函数：

.master()：设置运行模式，即：本地模式还是yarn模式appName()：顾名思义，设置名字enableHiveSupport() ：这是hive接口函数，如果想要查询hive的表就需要执行这个函数getOrCreate(：这是最终的函数，创建或者获取。

然后如果是同一个数据库拿这个实例直接执行sql就可以了。如果是不同数据库那就需要像我这么写了，spark.write().jdbc()
spark目前好像是与hive和mysql接口集成，即如果读写hive和mysql是不需要jdbc的方式的，其他的都需要。jdbc有四个参数，url table。mode(‘append’,‘overwrite’,‘error’,等) ，properties (连接的配置，以字典形式传入，其中‘driver’项是驱动，是固定的，连value一起都是固定传入，其他参数可以酌情使用),
spark还有其他函数，这里就不一一介绍了。

再说说踩过的坑，一开始配置好环境，环境配置可以参考
在pycharm中安装spark环境
Hadoop安装教程 Mac版
这两个结合着看就行

坑点1:

报错
py4j.protocol.Py4JJavaError: An error occurred while calling o75.jdbc.
java.lang.ClassNotFoundException:ru.yandex.clickhouse.ClickHouseDriver
没有clickhouse的驱动，去下载一个驱动，放进spark下的jars中解决。

坑点2:可以读取，不可写入

这里耽误了好久，因为觉得已经能够读取了，那就肯定不是驱动的事，一直在找语法上的错误，最后在网上查到缺少guava-28.0-jre.jar包
错误信息为：
java.lang.NoClassDefFoundError: com/google/common/escape/Escapers
去下载一个放进 spark中的jar下，解决，但是要注意，3.1spark自带一个guava-14.0.jar,但自带的这个jar版本不够，必须放进去更高的版本。测试几次好像最低要guava-16.jar才行。另外，网上有人遇到放进spark的jars路径下无法解决，可以试试其他路径，参考链接：
Caused by: java.lang.NoClassDefFoundError: com/google/common/escape/Escapers

坑点3:不能overwrite

这个坑目前未解决，只要我用overwrite模式，就会报错，错误信息很粗略，改用append模式就能够成功写入，现在采取的方案是在执行spark程序之前先用clickhouse_driver 进行清空表的操作，然后使用append的模式写入。另外说一下，好像好多etl工具的overwrite模式好像都不怎么靠谱，动不动就报错。

最后，成功写入

补充：在pycharm中添加spark的环境变量：
和上面文档中提示添加的路径变量没有任何关系，重新梳理一下pycarm中的配置步骤。

创建项目，指定python解释器的版本

配置项目依赖，

py4j 将Python代码转换为Java代码的库
pyspark Python的Spark编程依赖库

选择Add Content Root，在弹出的文件选择框中，选择Spark安装目录中的python文件夹下的lib目录中的py4j和pyspark的依赖文件，点击OK，将这两个依赖加入到当前项目的依赖库中。

点击右上角的程序配置，配置环境变量，添加SPARK_HOME值为SPARK安装包的解压路径，PYSAPRK_PYTHON指定PySpark的Python命令，注意python版本为3.6+。

参考文档
将数据通过spark从hive导入到Clickhouse

pyspark从hive导出数据到clickhouse(pyspark杩炴帴hive)

大数据系统相关栏目本月热门文章