0
0

六十三、Spark-读取数据并写入数据库

支持的数据源-JDBC

需求说明：使用Spark流式计算 将数据写入MySQL，并读取数据库信息进行打印

项目主体架构

pom.xml依赖

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <modelVersion>4.0.0</modelVersion>
  7.     <groupId>cn.itcast</groupId>
  8.     <artifactId>SparkDemo</artifactId>
  9.     <version>1.0-SNAPSHOT</version>
  11.     <repositories>
  12.         <repository>
  13.             <id>aliyun</id>
  14.             <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
  15.         </repository>
  16.         <repository>
  17.             <id>apache</id>
  18.             <url>https://repository.apache.org/content/repositories/snapshots/</url>
  19.         </repository>
  20.         <repository>
  21.             <id>cloudera</id>
  22.             <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
  23.         </repository>
  24.     </repositories>
  25.     <properties>
  26.         <encoding>UTF-8</encoding>
  27.         <maven.compiler.source>1.8</maven.compiler.source>
  28.         <maven.compiler.target>1.8</maven.compiler.target>
  29.         <scala.version>2.12.11</scala.version>
  30.         <spark.version>3.0.1</spark.version>
  31.         <hadoop.version>2.7.5</hadoop.version>
  32.     </properties>
  33.     <dependencies>
  34.         <!--依赖Scala语言-->
  35.         <dependency>
  36.             <groupId>org.scala-lang</groupId>
  37.             <artifactId>scala-library</artifactId>
  38.             <version>${scala.version}</version>
  39.         </dependency>
  41.         <!--SparkCore依赖-->
  42.         <dependency>
  43.             <groupId>org.apache.spark</groupId>
  44.             <artifactId>spark-core_2.12</artifactId>
  45.             <version>${spark.version}</version>
  46.         </dependency>
  48.         <!-- spark-streaming-->
  49.         <dependency>
  50.             <groupId>org.apache.spark</groupId>
  51.             <artifactId>spark-streaming_2.12</artifactId>
  52.             <version>${spark.version}</version>
  53.         </dependency>
  55.         <!--spark-streaming+Kafka依赖-->
  56.         <dependency>
  57.             <groupId>org.apache.spark</groupId>
  58.             <artifactId>spark-streaming-kafka-0-10_2.12</artifactId>
  59.             <version>${spark.version}</version>
  60.         </dependency>
  62.         <!--SparkSQL依赖-->
  63.         <dependency>
  64.             <groupId>org.apache.spark</groupId>
  65.             <artifactId>spark-sql_2.12</artifactId>
  66.             <version>${spark.version}</version>
  67.         </dependency>
  69.         <!--SparkSQL+ Hive依赖-->
  70.         <dependency>
  71.             <groupId>org.apache.spark</groupId>
  72.             <artifactId>spark-hive_2.12</artifactId>
  73.             <version>${spark.version}</version>
  74.         </dependency>
  75.         <dependency>
  76.             <groupId>org.apache.spark</groupId>
  77.             <artifactId>spark-hive-thriftserver_2.12</artifactId>
  78.             <version>${spark.version}</version>
  79.         </dependency>
  81.         <!--StructuredStreaming+Kafka依赖-->
  82.         <dependency>
  83.             <groupId>org.apache.spark</groupId>
  84.             <artifactId>spark-sql-kafka-0-10_2.12</artifactId>
  85.             <version>${spark.version}</version>
  86.         </dependency>
  88.         <!-- SparkMlLib机器学习模块,里面有ALS推荐算法-->
  89.         <dependency>
  90.             <groupId>org.apache.spark</groupId>
  91.             <artifactId>spark-mllib_2.12</artifactId>
  92.             <version>${spark.version}</version>
  93.         </dependency>
  95.         <dependency>
  96.             <groupId>org.apache.hadoop</groupId>
  97.             <artifactId>hadoop-client</artifactId>
  98.             <version>2.7.5</version>
  99.         </dependency>
  101.         <dependency>
  102.             <groupId>com.hankcs</groupId>
  103.             <artifactId>hanlp</artifactId>
  104.             <version>portable-1.7.7</version>
  105.         </dependency>
  107.         <dependency>
  108.             <groupId>mysql</groupId>
  109.             <artifactId>mysql-connector-java</artifactId>
  110.             <version>8.0.23</version>
  111.         </dependency>
  113.         <dependency>
  114.             <groupId>redis.clients</groupId>
  115.             <artifactId>jedis</artifactId>
  116.             <version>2.9.0</version>
  117.         </dependency>
  119.         <dependency>
  120.             <groupId>com.alibaba</groupId>
  121.             <artifactId>fastjson</artifactId>
  122.             <version>1.2.47</version>
  123.         </dependency>
  125.         <dependency>
  126.             <groupId>org.projectlombok</groupId>
  127.             <artifactId>lombok</artifactId>
  128.             <version>1.18.2</version>
  129.             <scope>provided</scope>
  130.         </dependency>
  131.     </dependencies>
  133.     <build>
  134.         <sourceDirectory>src/main/scala</sourceDirectory>
  135.         <plugins>
  136.             <!-- 指定编译java的插件 -->
  137.             <plugin>
  138.                 <groupId>org.apache.maven.plugins</groupId>
  139.                 <artifactId>maven-compiler-plugin</artifactId>
  140.                 <version>3.5.1</version>
  141.             </plugin>
  142.             <!-- 指定编译scala的插件 -->
  143.             <plugin>
  144.                 <groupId>net.alchim31.maven</groupId>
  145.                 <artifactId>scala-maven-plugin</artifactId>
  146.                 <version>3.2.2</version>
  147.                 <executions>
  148.                     <execution>
  149.                         <goals>
  150.                             <goal>compile</goal>
  151.                             <goal>testCompile</goal>
  152.                         </goals>
  153.                         <configuration>
  154.                             <args>
  155.                                 <arg>-dependencyfile</arg>
  156.                                 <arg>${project.build.directory}/.scala_dependencies</arg>
  157.                             </args>
  158.                         </configuration>
  159.                     </execution>
  160.                 </executions>
  161.             </plugin>
  162.             <plugin>
  163.                 <groupId>org.apache.maven.plugins</groupId>
  164.                 <artifactId>maven-surefire-plugin</artifactId>
  165.                 <version>2.18.1</version>
  166.                 <configuration>
  167.                     <useFile>false</useFile>
  168.                     <disableXmlReport>true</disableXmlReport>
  169.                     <includes>
  170.                         <include>**/*Test.*</include>
  171.                         <include>**/*Suite.*</include>
  172.                     </includes>
  173.                 </configuration>
  174.             </plugin>
  175.             <plugin>
  176.                 <groupId>org.apache.maven.plugins</groupId>
  177.                 <artifactId>maven-shade-plugin</artifactId>
  178.                 <version>2.3</version>
  179.                 <executions>
  180.                     <execution>
  181.                         <phase>package</phase>
  182.                         <goals>
  183.                             <goal>shade</goal>
  184.                         </goals>
  185.                         <configuration>
  186.                             <filters>
  187.                                 <filter>
  188.                                     <artifact>*:*</artifact>
  189.                                     <excludes>
  190.                                         <exclude>META-INF/*.SF</exclude>
  191.                                         <exclude>META-INF/*.DSA</exclude>
  192.                                         <exclude>META-INF/*.RSA</exclude>
  193.                                     </excludes>
  194.                                 </filter>
  195.                             </filters>
  196.                             <transformers>
  197.                                 <transformer
  198.                                         implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
  199.                                     <mainClass></mainClass>
  200.                                 </transformer>
  201.                             </transformers>
  202.                         </configuration>
  203.                     </execution>
  204.                 </executions>
  205.             </plugin>
  206.         </plugins>
  207.     </build>
  208. </project>
  1.     注：pom依赖在业务实施中是极其重要的一环，相当于配置文件，例如可能需要的 jar 包，可能需要的 Scala 语言版本都在此处进行配置 等等

创建数据库

  1. CREATE TABLE `data` (
  2.   `id` int(11) NOT NULL AUTO_INCREMENT,
  3.   `name` varchar(255) DEFAULT NULL,
  4.   `age` int(11) DEFAULT NULL,
  5.   PRIMARY KEY (`id`)
  6. ) ENGINE=InnoDB DEFAULT CHARSET=utf8;

业务逻辑

1、创建本地环境，并设置日志提示级别

  1. val conf: SparkConf = new SparkConf().setAppName("spark").setMaster("local[*]")
  2. val sc: SparkContext = new SparkContext(conf)
  3. sc.setLogLevel("WARN")

2、加载数据，创建RDD

  1. val dataRDD: RDD[(String, Int)] = sc.makeRDD(List(("tuomasi", 21), ("孙悟空", 19), ("猪八戒", 20)))

3、分区迭代

  1. dataRDD.foreachPartition(iter => {
  2. })

4、加载驱动

  1. val conn: Connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/bigdata?characterEncoding=UTF-8", "root", "123456")

5、封装SQL语句

  1. val sql: String = "INSERT INTO `data` (`id`, `name`, `age`) VALUES (NULL, ?, ?);"
  3. val ps: PreparedStatement = conn.prepareStatement(sql)

6、数据处理

  1. iter.foreach(t => { //t就表示每一条数据
  2. val name: String = t._1
  3. val age: Int = t._2
  4. ps.setString(1, name)
  5. ps.setInt(2, age)
  6. ps.addBatch()
  7. })
  8. ps.executeBatch()

7、关闭连接

  1. if (conn != null) conn.close()
  2. if (ps != null) ps.close()

8、读取数据库

  1. val getConnection = () => DriverManager.getConnection("jdbc:mysql://localhost:3306/bigdata?characterEncoding=UTF-8", "root", "123456")

9、SQL语句上下界设定以及分区数设置

  1. val studentTupleRDD: JdbcRDD[(Int, String, Int)] = new JdbcRDD[(Int, String, Int)](
  2.       sc,
  3.       getConnection,
  4.       sql,
  5.       1,      //id为1~20之间的记录进行提取
  6.       20,
  7.       1,
  8.       mapRow
  9.     )

10、结果集处理函数

  1. val mapRow: ResultSet => (Int, String, Int) = (r: ResultSet) => {
  2.       val id: Int = r.getInt("id")
  3.       val name: String = r.getString("name")
  4.       val age: Int = r.getInt("age")
  5.       (id, name, age)
  6.     }

11、遍历打印数据

  1. studentTupleRDD.foreach(println)

完整代码

  1. package org.example.spark
  3. import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}
  5. import org.apache.spark.rdd.{JdbcRDD, RDD}
  6. import org.apache.spark.{SparkConf, SparkContext}
  8. object RDD_DataSource {
  9.   def main(args: Array[String]): Unit = {
  10.     //TODO 0.env/创建环境
  11.     val conf: SparkConf = new SparkConf().setAppName("spark").setMaster("local[*]")
  12.     val sc: SparkContext = new SparkContext(conf)
  13.     sc.setLogLevel("WARN")
  15.     //TODO 1.source/加载数据/创建RDD
  16.     //RDD[(姓名, 年龄)]
  17.     val dataRDD: RDD[(String, Int)] = sc.makeRDD(List(("tuomasi", 21), ("孙悟空", 19), ("猪八戒", 20)))
  19.     //TODO 2.transformation
  20.     //TODO 3.sink/输出
  21.     //需求:将数据写入到MySQL,再从MySQL读出来
  22.     dataRDD.foreachPartition(iter => {
  23.       //加载驱动
  24.       val conn: Connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/bigdata?characterEncoding=UTF-8", "root", "123456")
  26.       val sql: String = "INSERT INTO `data` (`id`, `name`, `age`) VALUES (NULL, ?, ?);"
  28.       val ps: PreparedStatement = conn.prepareStatement(sql)
  30.       iter.foreach(t => { //t就表示每一条数据
  31.         val name: String = t._1
  32.         val age: Int = t._2
  33.         ps.setString(1, name)
  34.         ps.setInt(2, age)
  35.         ps.addBatch()
  36.         //ps.executeUpdate()
  37.       })
  38.       ps.executeBatch()
  39.       //关闭连接
  40.       if (conn != null) conn.close()
  41.       if (ps != null) ps.close()
  42.     })
  44.     //    //从MySQL读取
  45.     val getConnection = () => DriverManager.getConnection("jdbc:mysql://localhost:3306/bigdata?characterEncoding=UTF-8", "root", "123456")
  46.     val sql: String = "select id,name,age from data where id >= ? and id <= ?"
  47.     val mapRow: ResultSet => (Int, String, Int) = (r: ResultSet) => {
  48.       val id: Int = r.getInt("id")
  49.       val name: String = r.getString("name")
  50.       val age: Int = r.getInt("age")
  51.       (id, name, age)
  52.     }
  53.     val studentTupleRDD: JdbcRDD[(Int, String, Int)] = new JdbcRDD[(Int, String, Int)](
  54.       sc,
  55.       getConnection,
  56.       sql,
  57.       1,
  58.       20,
  59.       1,
  60.       mapRow
  61.     )
  62.     studentTupleRDD.foreach(println)
  63.   }
  64. }

程序运行

控制台打印

** 数据库查看**

  1.      注：此为实验案例，在真实的场景中往往数据都是数以万计级别或者更多，优秀的代码往往体现在数据量极大的场景下，调优不失为一种升职加薪的必备技能

项目总结

  1.     总结：在代码编写过程中，难免出现知识匮乏，在遇到问题时，养成多看源码的好习惯，在以后的开发书写过程中会有事半功倍的效果，当然日志，及其 debug 的作用在开发中也不容小觑。
标签： spark big data 运维
本文转载自: https://blog.csdn.net/m0_54925305/article/details/122579557
版权归原作者 托马斯-酷涛 所有， 如有侵权，请联系我们删除。
登录后发布评论

“六十三、Spark-读取数据并写入数据库”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？