Spark _24 _读取JDBC中的数据创建DataFrame/DataSet(MySql为例)（三）

两种方式创建DataSet

现在数据库中创建表不能给插入少量数据。

javaapi:

package SparkSql;import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrameReader;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;import java.util.HashMap;
import java.util.Map;/*** @author George* @description* 读取JDBC中的数据创建DataFrame(MySql为例)* 两种方式创建DataFrame**/
public class JDBCDemo {public static void main(String[] args) {SparkConf conf = new SparkConf();conf.setAppName("jdbc");conf.setMaster("local");JavaSparkContext sc = new JavaSparkContext(conf);sc.setLogLevel("error");SQLContext sqlContext = new SQLContext(sc);/*** 第一种方式读取MySql数据库表，加载为DataFrame*/Map<String,String> options = new HashMap<>();options.put("url","jdbc:mysql://localhost:3306/spark");options.put("driver","com.mysql.jdbc.Driver");options.put("user","root");options.put("password","123456");options.put("dbtable","person");// options   为基础数据源添加输入选项。Dataset<Row> person = sqlContext.read().format("jdbc").options(options).load();person.show();/*** +---+----------+---+* | id|      name|age|* +---+----------+---+* |  1|    George| 22|* |  2|  kangkang| 20|* |  3|GeorgeDage| 28|* |  4|    limumu|  1|* +---+----------+---+*/person.registerTempTable("person");/*** 第二种方式读取MySql数据表加载为DataFrame*/DataFrameReader reader = sqlContext.read().format("jdbc");reader.option("url","jdbc:mysql://localhost:3306/spark");reader.option("driver","com.mysql.jdbc.Driver");reader.option("user","root");reader.option("password","123456");reader.option("dbtable","score");Dataset<Row> load = reader.load();load.show();/*** +---+----------+-----+* | id|      name|score|* +---+----------+-----+* |  1|    George|  100|* |  2|  kangkang|  100|* |  3|GeorgeDage|   90|* |  4|    limumu|  120|* +---+----------+-----+*/load.registerTempTable("score");sc.stop();}
}

scalaAPI:

并将组合的数据重新插入到mysql中

package SparkSqlimport java.util.Propertiesimport org.apache.spark.sql.{SQLContext, SaveMode}
import org.apache.spark.{SparkConf, SparkContext}import scala.collection.mutableobject JDBCScalaDemo {def main(args: Array[String]): Unit = {val conf = new SparkConf().setAppName("jdbc").setMaster("local")val sc = new SparkContext(conf)sc.setLogLevel("error")val sqlContext = new SQLContext(sc)/*** 第一种方式读取Mysql数据库表创建DF*/val options = new mutable.HashMap[String,String]()options.put("url", "jdbc:mysql://localhost:3306/spark")options.put("driver","com.mysql.jdbc.Driver")options.put("user","root")options.put("password", "123456")options.put("dbtable","person")val person = sqlContext.read.format("jdbc").options(options).load()person.show()person.registerTempTable("person")/*** 第二种方式读取Mysql数据库表创建DF*/val reader = sqlContext.read.format("jdbc")reader.option("url", "jdbc:mysql://localhost:3306/spark")reader.option("driver","com.mysql.jdbc.Driver")reader.option("user","root")reader.option("password","123456")reader.option("dbtable", "score")val frame = reader.load()frame.show()frame.registerTempTable("score")/*** +---+----------+---+* | id|      name|age|* +---+----------+---+* |  1|    George| 22|* |  2|  kangkang| 20|* |  3|GeorgeDage| 28|* |  4|    limumu|  1|* +---+----------+---+** +---+----------+-----+* | id|      name|score|* +---+----------+-----+* |  1|    George|  100|* |  2|  kangkang|  100|* |  3|GeorgeDage|   90|* |  4|    limumu|  120|* +---+----------+-----+*/val result = sqlContext.sql("select person.id,person.name,person.age," +"score.score from person,score where person.id = score.id")result.show()/*** +---+----------+---+-----+* | id|      name|age|score|* +---+----------+---+-----+* |  1|    George| 22|  100|* |  3|GeorgeDage| 28|   90|* |  4|    limumu|  1|  120|* |  2|  kangkang| 20|  100|* +---+----------+---+-----+*//*** 将数据写入到mysql中*/val properties = new Properties()properties.setProperty("user", "root")properties.setProperty("password", "123456")result.write.mode(SaveMode.Append).jdbc("jdbc:mysql://localhost:3306/spark", "result", properties)sc.stop()}
}

结果：去mysql查看：

Spark _24 _读取JDBC中的数据创建DataFrame/DataSet(MySql为例)（三）相关推荐

Spark _25 _读取Hive中的数据加载成DataFrame/DataSet（四）
由于Hive不在本地,操作略显麻烦.不过细心一点,分析错误,也还好,如果你搭建的hadoop是HA,需要多注意: 这里指出一个错误,如果你报了同类错误,可以参考:https://georgedage. ...
Spark _25.plus _使用idea读取Hive中的数据加载成DataFrame/DataSet（四）
对Spark _25 _读取Hive中的数据加载成DataFrame/DataSet(四) https://georgedage.blog.csdn.net/article/details/10309 ...
Spark读取MySQL中的数据为DataFrame
Spark读取JDBC中的数据(以MySQL为例)为DataFrame,有两种方式. //聚合的时候默认分区是200,可以在此设置 val spark = SparkSession.builder() ...
Spark读取Hive中的数据加载为DataFrame
首先要告诉SparkSql,Hive在哪.然后读取Hive中的数据,必须开启enableHiveSupport. val spark = SparkSession.builder().appName( ...
python文件读取方法read(size)的含义是_在Python中可使用read([size])来读取文件中的数据,如果参数size省略,则读取文件中的()。...
[单选题]李明在他所属的公司工作五年,每天都很认真地处理繁杂的事情,同事们都夸他认真,但是依然没有建树,这是因为: [多选题]品牌标志的作用表现在 [单选题]新产品开发的第一个阶段是_______. ...
python文件读取方法read(size)的含义是_在Python中可使用read([size])来读取文件中的数据,如果参数size省略,则读取文件中的()。(4.0分)_学小易找答案...
[单选题]文本文件存储的是(),由若干文本行组成,通常每行以换行符 '\n' 结尾.(4.0分) [单选题]()属性是返回被打开文件的访问模式.(4.0分) [单选题]重力坝是由砼或( )修筑而成的大 ...
python开发的程序中以电子表格显示数据_使用 Python 读取电子表格中的数据实例详解...
Python 是最流行.功能最强大的编程语言之一.由于它是自由开源的,因此每个人都可以使用.大多数 Fedora 系统都已安装了该语言.Python 可用于多种任务,其中包括处理逗号分隔值(CSV)数 ...
StarRocks从入门到精通系列六：使用EXPORT、Spark、Flink从StarRocks中导出数据
StarRocks从入门到精通系列六:使用EXPORT.Spark.Flink从StarRocks中导出数据一.使用EXPORT导出数据 1.背景信息 2.导出流程 3.基本原理 4.相关配置 5. ...
使用jxl来读取Excel中的数据
使用java语言,借助第三方API来读取Excel中的数据 package com.excel;import java.io.File; import jxl.Cell; import jxl.She ...

Spark _24 _读取JDBC中的数据创建DataFrame/DataSet(MySql为例)（三）

Spark _24 _读取JDBC中的数据创建DataFrame/DataSet(MySql为例)（三）相关推荐

最新文章

热门文章