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。

相关文章

Debian系和Redhat系防火墙配置方式

Debian系和Redhat系防火墙配置方式

《Debian系和Redhat系防火墙配置方式》文章对比了Debian系UFW和Redhat系Firewalld防火墙的安装、启用禁用、端口管理、规则查看及注意事项,强调SSH端口需开放、规则持久化,... 目录Debian系UFW防火墙1. 安装2. 启用与禁用3. 基本命令4. 注意事项5. 示例配置R
阅读更多...
最新Spring Security的基于内存用户认证方式

最新Spring Security的基于内存用户认证方式

《最新SpringSecurity的基于内存用户认证方式》本文讲解SpringSecurity内存认证配置,适用于开发、测试等场景,通过代码创建用户及权限管理,支持密码加密,虽简单但不持久化,生产环... 目录1. 前言2. 因何选择内存认证？3. 基础配置实战❶ 创建Spring Security配置文件
阅读更多...
Python获取浏览器Cookies的四种方式小结

Python获取浏览器Cookies的四种方式小结

《Python获取浏览器Cookies的四种方式小结》在进行Web应用程序测试和开发时,获取浏览器Cookies是一项重要任务,本文我们介绍四种用Python获取浏览器Cookies的方式,具有一定的... 目录什么是 Cookie？1.使用Selenium库获取浏览器Cookies2.使用浏览器开发者工具
阅读更多...
Java获取当前时间String类型和Date类型方式

Java获取当前时间String类型和Date类型方式

《Java获取当前时间String类型和Date类型方式》：本文主要介绍Java获取当前时间String类型和Date类型方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,... 目录Java获取当前时间String和Date类型String类型和Date类型输出结果总结Java获取
阅读更多...
C#监听txt文档获取新数据方式

C#监听txt文档获取新数据方式

《C#监听txt文档获取新数据方式》文章介绍通过监听txt文件获取最新数据,并实现开机自启动、禁用窗口关闭按钮、阻止Ctrl+C中断及防止程序退出等功能,代码整合于主函数中,供参考学习... 目录前言一、监听txt文档增加数据二、其他功能1. 设置开机自启动2. 禁止控制台窗口关闭按钮3. 阻止Ctrl +
阅读更多...
详解MySQL中JSON数据类型用法及与传统JSON字符串对比

详解MySQL中JSON数据类型用法及与传统JSON字符串对比

《详解MySQL中JSON数据类型用法及与传统JSON字符串对比》MySQL从5.7版本开始引入了JSON数据类型,专门用于存储JSON格式的数据,本文将为大家简单介绍一下MySQL中JSON数据类型... 目录前言基本用法jsON数据类型 vs 传统JSON字符串1. 存储方式2. 查询方式对比3. 索引
阅读更多...
C#解析JSON数据全攻略指南

C#解析JSON数据全攻略指南

《C#解析JSON数据全攻略指南》这篇文章主要为大家详细介绍了使用C#解析JSON数据全攻略指南,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 目录一、为什么jsON是C#开发必修课？二、四步搞定网络JSON数据1. 获取数据 - HttpClient最佳实践2. 动态解析 - 快速
阅读更多...
linux批量替换文件内容的实现方式

linux批量替换文件内容的实现方式

《linux批量替换文件内容的实现方式》本文总结了Linux中批量替换文件内容的几种方法,包括使用sed替换文件夹内所有文件、单个文件内容及逐行字符串,强调使用反引号和绝对路径,并分享个人经验供参考... 目录一、linux批量替换文件内容 二、替换文件内所有匹配的字符串 三、替换每一行中全部str1为st
阅读更多...
MySQL 8 中的一个强大功能 JSON_TABLE示例详解

MySQL 8 中的一个强大功能 JSON_TABLE示例详解

《MySQL8中的一个强大功能JSON_TABLE示例详解》JSON_TABLE是MySQL8中引入的一个强大功能,它允许用户将JSON数据转换为关系表格式,从而可以更方便地在SQL查询中处理J... 目录基本语法示例示例查询解释应用场景不适用场景1. ‌jsON 数据结构过于复杂或动态变化‌2. ‌性能要
阅读更多...
Python实现终端清屏的几种方式详解

Python实现终端清屏的几种方式详解

《Python实现终端清屏的几种方式详解》在使用Python进行终端交互式编程时,我们经常需要清空当前终端屏幕的内容,本文为大家整理了几种常见的实现方法,有需要的小伙伴可以参考下... 目录方法一：使用 `os` 模块调用系统命令方法二：使用 `subprocess` 模块执行命令方法三：打印多个换行符模拟
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2