采用python开发sparkstreming全流程

最近开发了sparkstreaming的程序，且开发语言是采用python的，下述记录了开发的具体代码和过程，方便今后重复使用；

需要从kafka的topic上消费数据，最终写入到hadoop集群中，这里面有几个方案；
（1）采用kudu作为存储系统，直接将消费到的数据写入到kudu存储中，之后利用该数据;
（2）消费写入到文件中，放在hdfs上，采用hive-load的方式写入到hive表中，之后利用该数据；
（3）sparkstreaming直接写入到hive的分区表中，后续利用该数据；

注：方案（2）如果是流式处理的话，只能写到一个文件或文件夹当中，不好截取处理，我们最终采用了（3）方案，关于（3）方案会有小文件的问题，后续也会介绍该问题的解决方法；

我们采用python的pyspark开发了相应的代码，这里介绍一下具体的环境情况；
spark版本：2.4.0.7
pyspark的版本：2.4.6

代码如下：

#coding:utf-8

from pyspark.sql import SparkSession
 
# basic info
app_name = 'xxx'
 
# kafka info
kafka_broker_list = 'xxxx'
kafka_topic = 'xxx'
kafka_groupid = 'xxx'
kafka_username = 'xxx'
kafka_password = 'xxx'
 
def create_spark_session(app_name,log_level):
    '''
    create sparkSession,setting log level
    :param app_name:
    :return:
    '''
    spark_session = SparkSession 
        .builder 
        .appName(app_name) 
        .enableHiveSupport() 
        .getOrCreate()
    spark_session.sparkContext.setLogLevel('ERROR')
    return spark_session
          
def foreach_batch_function(batch_df, batch_id):
    '''
    '''
    # write no partition table
    #batch_df.write.format('hive').mode('append').saveAsTable('xxxx')
    # write partition table
    batch_df.write.format('hive').mode('append').partitionBy('partitionxxx').saveAsTable('xxx')
     
sqlstr1='''
create table xxx(xxx) partitioned by 'xxx' stored as parquet location 'hdfs://tmp/xxx'
'''       
def main():   
    # 1.create sparkSession
   spark_session = create_spark_session(app_name, 'ERROR')
 
    # 2.create table,this can not created by hive user，privilege permission
    #spark_session.sql(sqlstr1)
 
    # 3.read kafka data, using structstreaming
    df = spark_session.readStream 
        .format("kafka") 
        .option("kafka.bootstrap.servers", kafka_broker_list) 
        .option("subscribe", kafka_topic) 
        .option("kafka.sasl.mechanism", "PLAIN") 
        .option("kafka.security.protocol", "SASL_PLAINTEXT") 
        .option("kafka.sasl.jaas.config",
                'org.apache.kafka.common.security.plain.PlainLoginModule required username="xxx" password="xxx";') 
        .option("startingOffsets", "earliest") 
        .load()
     
    # 4. write partition table, you must set this config   
    spark_session.sql("set hive.exec.dynamic.partition.mode=nonstrict")
     
    # 5.get the data you want to，using sql
    words = df.selectExpr("cast(value as STRING) AS value")
    job_info_df = words.where("get_json_object(value,'$.xxx')='xxx'") 
                       .selectExpr("get_json_object(value,'$.xxx1') as xxx1",
                                   "get_json_object(value,'$.xxx2') as xxx2"
                       )
 
    # 5.using streaming, write table, split streaming to batch, write batch to table
    query1 = job_info_df. 
        writeStream. 
        foreachBatch(foreach_batch_function). 
        trigger(processingTime='1 minute'). 
        start()
    query1.awaitTermination()
    query1.stop()
 
     
if __name__ == '__main__':
    main()

注:
(1)我们在spark集群上运行python代码时，发现如果存在中文字符，会导致执行不通过，故建议不要使用中文注释，防止执行时报错；
(2)本代码是使用structstreaming的方式消费kafka的数据，经过sql处理，选取想要的数据，最终将数据写入到hive分区表中，在写入hive分区表时，我们通过foreachBatch的函数，将读取的流分成多个dataFrame，然后用对应的foreach_batch_function这个函数分别对dataFrame进行处理，这里使用了saveAsTable的方式写入到最终表中，表可以是分区表和非分区表；
(3)不知道是何原因，我们用hive直接建立最终表时，总是会报写入权限的问题，因此我们在sparksql中建立了最终表，且建立的是外表，目录是当前执行用户可以访问的目录；

代码执行采用spark-submit的方式执行，具体如下:

spark-submit --master yarn --keytab /home/xxx/xxx.keytab --principal xxx xxx.py

注:我们的集群采用kerberos认证的方式，因此提交时需要使用配置–kertab和–principal这两个参数；此外，我们在用cluster模式运行时，会导致报错，因此我们暂时采用client的模式运行python代码；

在用spark写入hive时，会出现很多的小文件，我们提供如下的解决方案：
（1）写入的表最好是分区表；
（2）对历史分区（非当前写入的分区）采用如下的命令解决小文件的问题；

INSERT OVERWRITE TABLE tablename PARTITION (partition)
SELECT * FROM tablename
DISTRIBUTE BY partition,cast(rand() * N as int)