pyspark 环境搭建和相关操作redis ,es

'''
pip3 install pyspark
pip3 install redis-py-cluster==2.1.3 -i http://pypi.douban.com/simple/ --trusted-host pypi.douban.com
'''
from pyspark.sql import SparkSession
from rediscluster import RedisCluster

redis_hosts = [
    {"host": "192.168.2.150", "port": 6379},
    {"host": "192.168.2.150", "port": 6380},
    {"host": "192.168.2.150", "port": 6381},
    {"host": "192.168.2.150", "port": 6382}
]
def write_hive2redis(df):
    json_rdd = df.rdd.map(lambda row:row.asDict())
    def write2redis(partition_data_list):
        redis_conn = RedisCluster(startup_nodes=redis_hosts, password="", decode_responses=True)
        for dic in partition_data_list:
            id = dic.get("id")
            name = dic.get("name")
            # 往redis中写数据
            redis_conn.set(id,name)
    json_rdd.foreachPartition(write2redis)

            
        
if __name__ == '__main__':
    # 创建SparkSession对象
    spark = SparkSession.builder.appName("HiveExample").enableHiveSupport().getOrCreate()
    sql="select * from table1"
    df = spark.sql(sql)
    write_hive2redis(df)
    spark.stop()