arrylist spark_java spark list 转为 RDD 转为 dataset 写入表中

古城微笑少年丶 2023-01-09 14:21 49阅读 0赞

package com.example.demo;

import java.util.ArrayList;

import java.util.Arrays;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import org.apache.spark.api.java.JavaRDD;

import org.apache.spark.api.java.JavaSparkContext;

import org.apache.spark.sql.Dataset;

import org.apache.spark.sql.Row;

import org.apache.spark.sql.SQLContext;

import org.apache.spark.sql.SparkSession;

public class DemoApplication \{

public static void main(String\[\] args) \{

///\*-----------------------线上调用方式--------------------------\*/

// 读入店铺id数据

SparkSession spark = SparkSession.builder().appName("demo\_spark").enableHiveSupport().getOrCreate();

Dataset 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");

System.out.println( "数据读取 OK" );

JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());

//JavaSparkContext sc = new JavaSparkContext();

SQLContext sqlContext = new SQLContext(sc);

// 将数据去重，转换成 List 格式

vender\_set = vender\_set.distinct();

vender\_set = vender\_set.na().fill(0L);

JavaRDD vender= vender\_set.toJavaRDD();

List vender\_list = vender.collect();

// 遍历商家id，调用jsf接口,创建list 保存返回数据

List list\_temp = new ArrayList();

for(Row row:vender\_list) \{

String id = row.getString(0);

String result = service.venderDownAmountList(id);

System.out.println( "接口调用返回值 OK" );

// 解析json串 ,按照JSONObject 和 JSONArray 一层一层解析 并过返回滤数据

JSONObject jsonOBJ = JSON.parseObject(result);

JSONArray data = jsonOBJ.getJSONArray("data");

if (data != null) \{

JSONObject data\_all = data.getJSONObject(0);

double amount = data\_all.getDouble("jfDownAmount");

// 将商家id 和 倒挂金额存下来

list\_temp.add("\{\\"vender\_id\\":"+id+",\\"amount\\":"+amount+"\}");

\}

else \{

continue;

\}

System.out.println( "解析 OK" );

\}

// list 转为 RDD

JavaRDD venderRDD = sc.parallelize(list\_temp);

// 注册成表

Dataset vender\_table = sqlContext.read().json(venderRDD);

vender\_table.registerTempTable("vender");

System.out.println( "注册表 OK" );

// 写入数据库

spark.sql("insert overwrite table dev.dev\_jypt\_vender\_dropaway\_amount select vender.vender\_id,vender.amount from vender");

System.out.println( "写入数据表 OK" );

sc.stop();

System.out.println( "Hello World!" );

\}

\}

发表评论取消回复

表情：

评论列表（有 0 条评论，49人围观）

还没有评论，来说两句吧...

相关阅读

相关 arrylist spark_java spark list 转为 RDD 转为 dataset 写入表中

package com.example.demo; import java.util.ArrayList; import java.util.Arrays; import

古城微笑少年丶/ 2023年01月09日 14:21/ 0 赞/ 50 阅读

相关 Spark中RDD、DataFrame和DataSet的区别与联系

目录一、RDD、DataFrame和DataSet的定义二、RDD、DataFrame和DataSet的比较（1）Sp

阳光穿透心脏的1/2处/ 2022年12月26日 13:29/ 0 赞/ 208 阅读

相关 Spark之RDD、DataSet、DataFrame转换

前面已经给大家说过RDD、DataSet、DataFrame三者之间的关系以及区别，各自都有各自的优劣。RDD、DataFrame、Dataset三者有许多共性，有各自适用的场

偏执的太偏执、/ 2022年11月28日 00:52/ 0 赞/ 326 阅读

相关 kotlin将json转为list

一、代码实现 fun <T> jsonToList(jsonList: String): List<T> { return gson.from

「爱情、让人受尽委屈。」/ 2022年08月31日 01:49/ 0 赞/ 162 阅读

相关 java中把多个map转为list

package com.yanshu.utils; import java.util.ArrayList; import java.util.HashMap

浅浅的花香味﹌/ 2022年06月02日 12:06/ 0 赞/ 248 阅读

相关 Spark RDD,DataFrame和DataSet的区别

RDD：编译时类型安全（编译时就进行类型检查），采用面向对象的编程风格。需要序列化和反序列化（对结构和数据），增加GC（垃圾回收）性能开销。 DataFrame：引入了sch

妖狐艹你老母/ 2022年05月24日 07:15/ 0 赞/ 193 阅读

相关 Spark系列--SparkSQL(二)RDD、DataFrame、DataSet

前言在SparkCore中我们知道数据抽象是RDD，在SparkSQL中Spark为我们提供了两个新的抽象，分别是DataFrame和DataSet。他们和RDD有什么

一时失言乱红尘/ 2022年05月19日 05:16/ 0 赞/ 206 阅读

相关将List集合转为json

需要导入的jar包 json-lib-2.4-jdk15.jar [下载地址][Link 1] commons-httpclient-3.1.jar

╰+哭是因爲堅強的太久メ/ 2022年03月10日 02:20/ 0 赞/ 312 阅读

相关 C#将 List 转为 byte[]

实例代码 using System; using System.Collections.Generic; using System.IO;

逃离我推掉我的手/ 2022年02月17日 13:42/ 0 赞/ 277 阅读

相关 string字符串转为List

String str = "1,,2,3"可以转为List<Long>集合 List<Long> idList = Arrays.asList(ids.spli

蔚落/ 2022年01月29日 06:01/ 0 赞/ 1157 阅读