JDBC,CaseClass,JSON,Parquet和Schema五种方式创建DataFrame

2023-12-10 16:58
文章标签 json 创建 jdbc 方式 五种 schema dataframe parquet caseclass

本文主要是介绍JDBC,CaseClass,JSON,Parquet和Schema五种方式创建DataFrame，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

1.JDBC的方式创建DataFrame

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.DataFrameReader;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;public class JDBC2MySQL {public static void main(String[] args){SparkConf conf=new SparkConf();conf.setAppName("JDBC2MySQL").setMaster("local");JavaSparkContext sc=new JavaSparkContext(conf);SQLContext sqlContext=new SQLContext(sc);/*1.通过format("jdbc")的方式说明SparkSQL操作的数据来源是通过JDBC获得*JDBC后端一般是数据库，例如MySQL、Oracle等*2.通过DataFrameReader的option方法把要访问的数据库的信息传递进去*3.url:代表数据库的jdbc链接地址*4.datable：代表具体要链接哪个数据库*5.Driver部分是Spark SQL访问数据库的具体的驱动的完整包名和类名*6.关于JDBC的驱动的Jar，可以放在Spark的library目录，也可以在使用SparkSubmit的使用指定Jar(编码和打包的时候都不需要这个JDBC的Jar)* */DataFrameReader reader=sqlContext.read().format("jdbc");reader.option("url", "jdbc:mysql://SparkMaster:3306");reader.option("dbtable","dt_spark");reader.option("driver", "com.mysql.jdbc.Driver");reader.option("user", "root");reader.option("password", "123");DataFrame mysqlDataSourceDF=reader.load();reader.option("dbtable", "dthadoop");DataFrame DFFromMySQL=reader.load();Map<String, String> options = new HashMap<String, String>();options.put("url", "jdbc:mysql://SparkMaster:3306/testdb");options.put("dbtable", "student_infos");options.put("user", "root");options.put("password","123");DataFrame studentInfosDF=sqlContext.read().format("jdbc").options(options).load();options.put("dbtable", "student_scores");DataFrame studentScoresDF=sqlContext.read().format("jdbc").options(options).load();List<Row> listRow=studentScoresDF.javaRDD().collect();for(Row row:listRow){System.out.println(row);}	 }
}

2.Case Class的方式创建DataFrame 

import java.util.List;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.Row;
/** 使用反射的方式将RDD转化为DataFrame*/
public class CaseClassDataFrame {public static void main(String[] args) {SparkConf conf=new SparkConf().setAppName("RDD2DataFrame").setMaster("local");JavaSparkContext sc=new JavaSparkContext(conf);SQLContext sqlContext=new SQLContext(sc);//获取SQLContextJavaRDD<String> lines=sc.textFile("C://Users//Jason Shu//Desktop//persons.txt");JavaRDD<Person> persons=lines.map(new Function<String, Person>()/*RDD<String>变为RDD<Person>,泛型转换*/{public Person call(String line) throws Exception {String[] splited=line.split(" ");Person p =new Person();p.setId(Integer.valueOf(splited[0].trim()));p.setName(splited[1]);p.setAge(Integer.valueOf(splited[0].trim()));return p;}});DataFrame df= sqlContext.createDataFrame(persons, Person.class);//SQLContext变为DataFrame /*creatDataFrame第一个参数JavaRDD<?>rdd，第二个参数Class<?>beanClass*/df.registerTempTable("persons");//注册一张临时表DataFrame bigData=sqlContext.sql("select * from persons where age >=6");JavaRDD<Row> bigDataRDD=bigData.javaRDD();//DataFrame转换为RDDJavaRDD<Person> result=bigDataRDD.map(new Function<Row, Person>()/*DataFrame转换为RDD，这个地方由于bigDataRDD是RDD<Row>,result是RDD<Person>* 相当于是一个泛型转换*/ {public Person call(Row row) throws Exception {Person p =new Person();p.setId(row.getInt(0));p.setName(row.getString(1));p.setAge(row.getInt(2));return p;}});List<Person> personList=result.collect();for(Person p:personList){System.out.println(p);}   }}

Person类 

public class Person {private static final long serialVesionUID=1L;private int id;private String name;private int age;@Overridepublic String toString() {return "Person [id=" + id + ", name=" + name + ", age=" + age + "]";}public int getId() {return id;}public void setId(int id) {this.id = id;}public String getName() {return name;}public void setName(String name) {this.name = name;}public int getAge() {return age;}public void setAge(int age) {this.age = age;}	}


3.JSON方式创建DataFrame

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.DataFrame;
public class JSONDataFrame {public static void main(String[] args) {SparkConf conf =new SparkConf().setAppName("DataFrame").setMaster("spark://SparkMaster:7077");JavaSparkContext sc =new JavaSparkContext(conf);SQLContext sqlContext=new SQLContext(sc);//可以简单的认为DataFrame是一张表DataFrame dataFrame=sqlContext.read().json("hdfs://SparkMaster:9000/data/people.json");dataFrame.show();}}

4.Parquet的方式创建DataFrame

import java.util.List;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
public class ParquetDataFrame {public static void main(String[] args) {SparkConf conf=new SparkConf();conf.setAppName("ParquetDataFrame").setMaster("spark://SparkMaster:7077");JavaSparkContext sc=new JavaSparkContext(conf);SQLContext sqlContext=new SQLContext(sc);DataFrame df=sqlContext.read().parquet("/input/people.parquet");df.registerTempTable("users");DataFrame result=sqlContext.sql("select name from users");List<Row> listRow=result.javaRDD().collect();for(Row row:listRow){System.out.println(row);}}
}

5.Schema的方式创建RDD

import java.util.ArrayList;
import java.util.List;
import org.apache.hadoop.hive.serde2.objectinspector.StructField;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructType;public class SchemaDataFrame {public static void main(String[] args) {SparkConf conf =new SparkConf();conf.setAppName("SchemaDataFrame").setMaster("local");JavaSparkContext sc=new JavaSparkContext(conf);//如果是sparkcontext就不会textfile(location)，而是textfile(location,partition)SQLContext sqlContext=new SQLContext(sc);JavaRDD<String> lines=sc.textFile("C://Users//Jason Shu//Desktop");JavaRDD<Row> personsRDD=lines.map(new Function<String, Row>() //JavaRDD<String>变为JavaRDD<Row>{public Row call(String line) throws Exception {String[] splited=line.split(",");return RowFactory.create(Integer.valueOf(splited[0]),splited[1],Integer.valueOf(splited[2]));}	});List<StructField> structFields=new ArrayList<StructField>();//构造一个StructFieldstructFields.add((StructField) DataTypes.createStructField("id",DataTypes.IntegerType,true));structFields.add((StructField) DataTypes.createStructField("name",DataTypes.StringType,true));structFields.add((StructField) DataTypes.createStructField("age",DataTypes.IntegerType,true));StructType structType=DataTypes.createStructType(structFields); }}


这篇关于JDBC,CaseClass,JSON,Parquet和Schema五种方式创建DataFrame的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

原文地址:
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.chinasem.cn/article/477711。

相关文章

C++中零拷贝的多种实现方式

C++中零拷贝的多种实现方式

《C++中零拷贝的多种实现方式》本文主要介绍了C++中零拷贝的实现示例,旨在在减少数据在内存中的不必要复制,从而提高程序性能、降低内存使用并减少CPU消耗,零拷贝技术通过多种方式实现,下面就来了解一下... 目录一、C++中零拷贝技术的核心概念二、std::string_view 简介三、std::stri
阅读更多...
Linux脚本(shell)的使用方式

Linux脚本(shell)的使用方式

《Linux脚本(shell)的使用方式》：本文主要介绍Linux脚本(shell)的使用方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录概述语法详解数学运算表达式Shell变量变量分类环境变量Shell内部变量自定义变量：定义、赋值自定义变量：引用、修改、删
阅读更多...
Python中使用uv创建环境及原理举例详解

Python中使用uv创建环境及原理举例详解

《Python中使用uv创建环境及原理举例详解》uv是Astral团队开发的高性能Python工具,整合包管理、虚拟环境、Python版本控制等功能,：本文主要介绍Python中使用uv创建环境及... 目录一、uv工具简介核心特点：二、安装uv1. 通过pip安装2. 通过脚本安装验证安装：配置镜像源（可
阅读更多...
python判断文件是否存在常用的几种方式

python判断文件是否存在常用的几种方式

《python判断文件是否存在常用的几种方式》在Python中我们在读写文件之前,首先要做的事情就是判断文件是否存在,否则很容易发生错误的情况,：本文主要介绍python判断文件是否存在常用的几种... 目录1. 使用 os.path.exists()2. 使用 os.path.isfile()3. 使用
阅读更多...
SpringBoot排查和解决JSON解析错误(400 Bad Request)的方法

SpringBoot排查和解决JSON解析错误(400 Bad Request)的方法

《SpringBoot排查和解决JSON解析错误(400BadRequest)的方法》在开发SpringBootRESTfulAPI时,客户端与服务端的数据交互通常使用JSON格式,然而,JSON... 目录问题背景1. 问题描述2. 错误分析解决方案1. 手动重新输入jsON2. 使用工具清理JSON3.
阅读更多...
Springboot3+将ID转为JSON字符串的详细配置方案

Springboot3+将ID转为JSON字符串的详细配置方案

《Springboot3+将ID转为JSON字符串的详细配置方案》：本文主要介绍纯后端实现Long/BigIntegerID转为JSON字符串的详细配置方案,s基于SpringBoot3+和Spr... 目录1. 添加依赖2. 全局 Jackson 配置3. 精准控制（可选）4. OpenAPI (Spri
阅读更多...
Mybatis的分页实现方式

Mybatis的分页实现方式

《Mybatis的分页实现方式》MyBatis的分页实现方式主要有以下几种,每种方式适用于不同的场景,且在性能、灵活性和代码侵入性上有所差异,对Mybatis的分页实现方式感兴趣的朋友一起看看吧... 目录​1. 原生 SQL 分页（物理分页）​​2. RowBounds 分页（逻辑分页）​​3. Page
阅读更多...
MySQL JSON 查询中的对象与数组技巧及查询示例

MySQL JSON 查询中的对象与数组技巧及查询示例

《MySQLJSON查询中的对象与数组技巧及查询示例》MySQL中JSON对象和JSON数组查询的详细介绍及带有WHERE条件的查询示例,本文给大家介绍的非常详细,mysqljson查询示例相关知... 目录jsON 对象查询1. JSON_CONTAINS2. JSON_EXTRACT3. JSON_TA
阅读更多...
Java中实现线程的创建和启动的方法

Java中实现线程的创建和启动的方法

《Java中实现线程的创建和启动的方法》在Java中,实现线程的创建和启动是两个不同但紧密相关的概念,理解为什么要启动线程（调用start()方法）而非直接调用run()方法,是掌握多线程编程的关键,... 目录1. 线程的生命周期2. start() vs run() 的本质区别3. 为什么必须通过 st
阅读更多...
Linux链表操作方式

Linux链表操作方式

《Linux链表操作方式》：本文主要介绍Linux链表操作方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、链表基础概念与内核链表优势二、内核链表结构与宏解析三、内核链表的优点四、用户态链表示例五、双向循环链表在内核中的实现优势六、典型应用场景七、调试技巧与
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2