java spark list 转为 RDD 转为 dataset 写入表中

1package com.example.demo;
2
3import java.util.ArrayList;
4import java.util.Arrays;
5import java.util.HashMap;
6import java.util.List;
7import java.util.Map;
8
9import org.apache.spark.api.java.JavaRDD;
10import org.apache.spark.api.java.JavaSparkContext;
11import org.apache.spark.sql.Dataset;
12import org.apache.spark.sql.Row;
13import org.apache.spark.sql.SQLContext;
14import org.apache.spark.sql.SparkSession;
15
16public class DemoApplication {
17
18	public static void main(String[] args) {
19		
20		
21//		/*-----------------------线上调用方式--------------------------*/
22		// 读入店铺id数据
23		SparkSession spark = SparkSession.builder().appName("demo_spark").enableHiveSupport().getOrCreate();
24		Dataset<Row> vender_set = spark.sql("select pop_vender_id from app.app_sjzt_payout_apply_with_order where dt = '2019-08-05' and pop_vender_id is not null");
25		System.out.println( "数据读取 OK" );
26		
27		
28		JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
29//		JavaSparkContext sc = new JavaSparkContext();
30		SQLContext sqlContext = new SQLContext(sc);
31
32		// 将数据去重，转换成 List<Row> 格式
33		vender_set =  vender_set.distinct();
34		vender_set = vender_set.na().fill(0L);
35		JavaRDD<Row> vender= vender_set.toJavaRDD();
36		List<Row> vender_list = vender.collect();
37		
38
39		// 遍历商家id，调用jsf接口,创建list 保存返回数据
40		List<String> list_temp = new ArrayList<String>();
41		for(Row row:vender_list) {
42			String id = row.getString(0);
43			String result = service.venderDownAmountList(id);
44			
45			System.out.println( "接口调用返回值 OK" );
46			
47			// 解析json串 ,按照JSONObject 和 JSONArray 一层一层解析 并过返回滤数据
48			JSONObject jsonOBJ = JSON.parseObject(result);
49			JSONArray data = jsonOBJ.getJSONArray("data");
50			if (data != null) {
51				JSONObject data_all = data.getJSONObject(0);
52				double amount = data_all.getDouble("jfDownAmount");
53				// 将商家id 和 倒挂金额存下来
54				list_temp.add("{\"vender_id\":"+id+",\"amount\":"+amount+"}");
55			}
56			else {
57				continue;
58			}
59			
60			System.out.println( "解析 OK" );
61			
62		}
63		// list 转为 RDD 
64		JavaRDD<String> venderRDD = sc.parallelize(list_temp);
65		
66		// 注册成表
67		Dataset<Row> vender_table = sqlContext.read().json(venderRDD);
68		vender_table.registerTempTable("vender");
69		System.out.println( "注册表 OK" );
70		
71		// 写入数据库
72		spark.sql("insert overwrite table dev.dev_jypt_vender_dropaway_amount select vender.vender_id,vender.amount from vender");
73		System.out.println( "写入数据表 OK" );
74
75		sc.stop();		
76		System.out.println( "Hello World!" );
77		
78	}
79}