Spark UDF实践之json解析

落日映苍穹つ 2022-04-03 05:08 396阅读 0赞

## Spark UDF实践之json解析 ##

我们一般使用spark处理json字段时，通常使用schema来约束json的字段，但是json数据中也会有一些需要特殊处理的字段需要获取，那么我们就需要通过UDF来进行处理了。

下面解析一个json的数据做一个示例：

json数据源：

{"final_score":16, "risk_items":[{"item_id":3403925, "item_name":"7天内申请人在多个平台申请借款", "risk_level":"high", "group":"多平台借贷申请检测", "item_detail":{"discredit_times":null, "overdue_details":null, "platform_count":2, "court_details":null, "fraud_type":null, "platform_detail":["一般消费分期平台:1", "P2P网贷:1"], "high_risk_areas":null, "hit_list_datas":null, "frequency_detail_list":null}},{"item_id":3403927,"item_name":"1个月内申请人在多个平台申请借款","risk_level":"medium","group":"多平台借贷申请检测","item_detail":{"discredit_times":null,"overdue_details":null,"platform_count":2,"court_details":null,"fraud_type":null,"platform_detail":["一般消费分期平台:1","P2P网贷:1"],"high_risk_areas":null,"hit_list_datas":null,"frequency_detail_list":null}},{"item_id":3403929,"item_name":"3个月内申请人在多个平台申请借款","risk_level":"medium","group":"多平台借贷申请检测","item_detail":{"discredit_times":null,"overdue_details":null,"platform_count":2,"court_details":null,"fraud_type":null,"platform_detail":["一般消费分期平台:1","P2P网贷:1"],"high_risk_areas":null,"hit_list_datas":null,"frequency_detail_list":null}},{"item_id":3403931,"item_name":"6个月内申请人在多个平台申请借款","risk_level":"medium","group":"多平台借贷申请检测","item_detail":{"discredit_times":null,"overdue_details":null,"platform_count":2,"court_details":null,"fraud_type":null,"platform_detail":["一般消费分期平台:1","P2P网贷:1"],"high_risk_areas":null,"hit_list_datas":null,"frequency_detail_list":null}},{"item_id":3403935,"item_name":"18个月内申请人在多个平台申请借款","risk_level":"low","group":"多平台借贷申请检测","item_detail":{"discredit_times":null,"overdue_details":null,"platform_count":2,"court_details":null,"fraud_type":null,"platform_detail":["一般消费分期平台:1","P2P网贷:1"],"high_risk_areas":null,"hit_list_datas":null,"frequency_detail_list":null}},{"item_id":3403937,"item_name":"24个月内申请人在多个平台申请借款","risk_level":"low","group":"多平台借贷申请检测","item_detail":{"discredit_times":null,"overdue_details":null,"platform_count":2,"court_details":null,"fraud_type":null,"platform_detail":["一般消费分期平台:1","P2P网贷:1"],"high_risk_areas":null,"hit_list_datas":null,"frequency_detail_list":null}},{"item_id":3403939,"item_name":"60个月以上申请人在多个平台申请借款","risk_level":"low","group":"多平台借贷申请检测","item_detail":{"discredit_times":null,"overdue_details":null,"platform_count":2,"court_details":null,"fraud_type":null,"platform_detail":["一般消费分期平台:1","P2P网贷:1"],"high_risk_areas":null,"hit_list_datas":null,"frequency_detail_list":null}}],"final_decision":"Accept","report_time":1495377281000,"success":true,"report_id":"ER2017052122344113605405","apply_time":1495377281000}

我们需要解析出，item\_name分别为：7天内申请人在多个平台申请借款，1个月内申请人在多个平台申请借款，3个月内申请人在多个平台申请借款，6个月内申请人在多个平台申请借款，对应的platform\_count的值。

下面就直接上代码了：

import os
    
    from pyspark.sql import SparkSession
    from pyspark.sql.types import StructType, StructField, StringType
    
    if __name__ == '__main__':
    
        # 定义UDF函数，字段的获取规则
        def parse_1(risk_items, item_name):
            for i in risk_items:
                print(i)
                print(type(i))
                try:
                    if i.item_name == item_name:
                        return i.item_detail.platform_count
                except:
                    return ""
    
    
        # 解决一个python环境的bug，本地默认是python3，这里用的是python2.7版本
        os.environ['PYSPARK_PYTHON'] = '/usr/bin/python'
    
        spark = SparkSession \
            .builder \
            .appName("application") \
            .master("local") \
            .getOrCreate()
    
        # 注册UDF函数，sparksql函数名为td_parse1，定义的func名parse_1
        spark.udf.register("td_parse1", parse_1)
        
        # 读取json数据
        df = spark.read.json("1.json")
        
        # 创建临时表
        df.createOrReplaceTempView("tmp")
    
        # 定义sparksql
        resDf = spark.sql(
            """
            select
            final_score as td_final_score,
            td_parse1(risk_items,'7天内申请人在多个平台申请借款') as td_platform_count_7d,
            td_parse1(risk_items,'1个月内申请人在多个平台申请借款') as td_platform_count_1m,
            td_parse1(risk_items,'3个月内申请人在多个平台申请借款') as td_platform_count_3m,
            td_parse1(risk_items,'6个月内申请人在多个平台申请借款') as td_platform_count_6m
            from tmp""")
    
        # 展示数据
        resDf.show()
        spark.stop()

解析出来的结果如图：

![image][]

[image]: /images/20220403/04f7b01f7fc248b582ab89aa62c89405.png