我的书签
添加书签
移除书签使用 Spark 将批量数据写入 MatrixOne
概述
Apache Spark 是一个为高效处理大规模数据而设计的分布式计算引擎。它采用分布式并行计算的方式，将数据拆分、计算、合并的任务分散到多台计算机上，从而实现了高效的数据处理和分析。
应用场景
- 大规模数据处理与分析
Spark 能够处理海量数据，通过并行计算任务提高了处理效率。它广泛应用于金融、电信、医疗等领域的数据处理和分析。 - 流式数据处理
Spark Streaming 允许实时处理数据流，将其转化为可供分析和存储的批处理数据。这在在线广告、网络安全等实时数据分析场景中非常有用。 - 机器学习
Spark 提供了机器学习库（MLlib），支持多种机器学习算法和模型训练，用于推荐系统、图像识别等机器学习应用。 - 图计算
Spark 的图计算库（GraphX）支持多种图计算算法，适用于社交网络分析、推荐系统等图分析场景。
本篇文档将介绍两种使用 Spark 计算引擎实现批量数据写入 MatrixOne 的示例。一种示例是从 MySQL 迁移数据至 MatrixOne，另一种是将 Hive 数据写入 MatrixOne。
前期准备
硬件环境
本次实践对于机器的硬件要求如下：
| 服务器名称 | 服务器 IP | 安装软件 | 操作系统 |
|---|---|---|---|
| node1 | 192.168.146.10 | MatrixOne | Debian11.1 x86 |
| node3 | 192.168.146.11 | IDEA、MYSQL、Hadoop、Hive | Windows 10 |
软件环境
本次实践需要安装部署以下软件环境：
- 已完成安装和启动 MatrixOne。
- 下载并安装 IntelliJ IDEA version 2022.2.1 及以上。
- 下载并安装 JDK 8+。
- 如需从 Hive 导入数据，需要安装 Hadoop 和 Hive。
- 下载并安装 MySQL Client 8.0.33。
示例 1：从 MySQL 迁移数据至 MatrixOne
步骤一：初始化项目
-
启动 IDEA，点击 File > New > Project，选择 Spring Initializer，并填写以下配置参数：
-
Name：mo-spark-demo
- Location：~\Desktop
- Language：Java
- Type：Maven
- Group：com.example
- Artiface：matrixone-spark-demo
- Package name：com.matrixone.demo
- JDK 1.8
2. 添加项目依赖，在项目根目录下的 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>com.example.mo</groupId>
8. <artifactId>mo-spark-demo</artifactId>
9. <version>1.0-SNAPSHOT</version>
11. <properties>
12. <maven.compiler.source>8</maven.compiler.source>
13. <maven.compiler.target>8</maven.compiler.target>
14. <spark.version>3.2.1</spark.version>
15. </properties>
17. <dependencies>
18. <dependency>
19. <groupId>org.apache.spark</groupId>
20. <artifactId>spark-sql_2.12</artifactId>
21. <version>${spark.version}</version>
22. </dependency>
24. <dependency>
25. <groupId>org.apache.spark</groupId>
26. <artifactId>spark-hive_2.12</artifactId>
27. <version>${spark.version}</version>
28. </dependency>
29. <dependency>
30. <groupId>org.apache.spark</groupId>
31. <artifactId>spark-catalyst_2.12</artifactId>
32. <version>${spark.version}</version>
33. </dependency>
34. <dependency>
35. <groupId>org.apache.spark</groupId>
36. <artifactId>spark-core_2.12</artifactId>
37. <version>${spark.version}</version>
38. </dependency>
39. <dependency>
40. <groupId>org.codehaus.jackson</groupId>
41. <artifactId>jackson-core-asl</artifactId>
42. <version>1.9.13</version>
43. </dependency>
44. <dependency>
45. <groupId>org.codehaus.jackson</groupId>
46. <artifactId>jackson-mapper-asl</artifactId>
47. <version>1.9.13</version>
48. </dependency>
51. <dependency>
52. <groupId>mysql</groupId>
53. <artifactId>mysql-connector-java</artifactId>
54. <version>8.0.16</version>
55. </dependency>
57. </dependencies>
59. </project>
步骤二：读取 MatrixOne 数据
使用 MySQL 客户端连接 MatrixOne 后，创建演示所需的数据库以及数据表。
- 在 MatrixOne 中创建数据库、数据表，并导入数据：
1. CREATE DATABASE test;
2. USE test;
3. CREATE TABLE `person` (`id` INT DEFAULT NULL, `name` VARCHAR(255) DEFAULT NULL, `birthday` DATE DEFAULT NULL);
4. INSERT INTO test.person (id, name, birthday) VALUES(1, 'zhangsan', '2023-07-09'),(2, 'lisi', '2023-07-08'),(3, 'wangwu', '2023-07-12');
2. 在 IDEA 中创建 MoRead.java 类，以使用 Spark 读取 MatrixOne 数据：
``` 1. package com.matrixone.spark;
- import org.apache.spark.sql.Dataset;
- import org.apache.spark.sql.Row;
- import org.apache.spark.sql.SQLContext;
-
import org.apache.spark.sql.SparkSession;
-
import java.util.Properties;
-
/**
-
- @auther MatrixOne
-
- @desc 读取 MatrixOne 数据
- */
-
public class MoRead {
-
// parameters
- private static String master = "local[2]";
-
private static String appName = "mo_spark_demo";
-
private static String srcHost = "192.168.146.10";
- private static Integer srcPort = 6001;
- private static String srcUserName = "root";
- private static String srcPassword = "111";
- private static String srcDataBase = "test";
-
private static String srcTable = "person";
-
public static void main(String[] args) {
- SparkSession sparkSession = SparkSession.builder().appName(appName).master(master).getOrCreate();
- SQLContext sqlContext = new SQLContext(sparkSession);
- Properties properties = new Properties();
- properties.put("user", srcUserName);
- properties.put("password", srcPassword);
- Dataset
dataset = sqlContext.read()
- .jdbc("jdbc:mysql://" + srcHost + ":" + srcPort + "/" + srcDataBase,srcTable, properties);
- dataset.show();
-
}
-
} ```
- 在 IDEA 中运行
MoRead.Main()，执行结果如下：
步骤三：将 MySQL 数据写入 MatrixOne
现在可以开始使用 Spark 将 MySQL 数据迁移到 MatrixOne。
- 准备 MySQL 数据：在 node3 上，使用 Mysql 客户端连接本地 Mysql，创建所需数据库、数据表、并插入数据：
1. mysql -h127.0.0.1 -P3306 -uroot -proot
2. mysql> CREATE DATABASE motest;
3. mysql> USE motest;
4. mysql> CREATE TABLE `person` (`id` int DEFAULT NULL, `name` varchar(255) DEFAULT NULL, `birthday` date DEFAULT NULL);
5. mysql> INSERT INTO motest.person (id, name, birthday) VALUES(2, 'lisi', '2023-07-09'),(3, 'wangwu', '2023-07-13'),(4, 'zhaoliu', '2023-08-08');
2. 清空 MatrixOne 表数据：
在 node3 上，使用 MySQL 客户端连接本地 MatrixOne。由于本示例继续使用前面读取 MatrixOne 数据的示例中的 test 数据库，因此我们需要首先清空 person 表的数据。
1. -- 在 node3 上，使用 Mysql 客户端连接 node1 的 MatrixOne
2. mysql -h192.168.146.10 -P6001 -uroot -p111
3. mysql> TRUNCATE TABLE test.person;
3. 在 IDEA 中编写代码：
创建 Person.java 和 Mysql2Mo.java 类，使用 Spark 读取 MySQL 数据。Mysql2Mo.java 类代码可参考如下示例：
1. package com.matrixone.spark;
3. import org.apache.spark.api.java.function.MapFunction;
4. import org.apache.spark.sql.*;
6. import java.sql.SQLException;
7. import java.util.Properties;
9. /**
10. * @auther MatrixOne
11. * @desc
12. */
13. public class Mysql2Mo {
15. // parameters
16. private static String master = "local[2]";
17. private static String appName = "app_spark_demo";
19. private static String srcHost = "127.0.0.1";
20. private static Integer srcPort = 3306;
21. private static String srcUserName = "root";
22. private static String srcPassword = "root";
23. private static String srcDataBase = "motest";
24. private static String srcTable = "person";
26. private static String destHost = "192.168.146.10";
27. private static Integer destPort = 6001;
28. private static String destUserName = "root";
29. private static String destPassword = "111";
30. private static String destDataBase = "test";
31. private static String destTable = "person";
34. public static void main(String[] args) throws SQLException {
35. SparkSession sparkSession = SparkSession.builder().appName(appName).master(master).getOrCreate();
36. SQLContext sqlContext = new SQLContext(sparkSession);
37. Properties connectionProperties = new Properties();
38. connectionProperties.put("user", srcUserName);
39. connectionProperties.put("password", srcPassword);
40. connectionProperties.put("driver","com.mysql.cj.jdbc.Driver");
42. //jdbc.url=jdbc:mysql://127.0.0.1:3306/database
43. String url = "jdbc:mysql://" + srcHost + ":" + srcPort + "/" + srcDataBase + "?characterEncoding=utf-8&autoReconnect=true&zeroDateTimeBehavior=convertToNull&useSSL=false&serverTimezone=Asia/Shanghai";
45. //SparkJdbc 读取表内容
46. System.out.println("读取数据库中 person 的表内容");
47. // 读取表中所有数据
48. Dataset<Row> rowDataset = sqlContext.read().jdbc(url,srcTable,connectionProperties).select("*");
49. //显示数据
50. //rowDataset.show();
51. //筛选 id > 2 的数据，并将 name 字段添加 spark_ 前缀
52. Dataset<Row> dataset = rowDataset.filter("id > 2")
53. .map((MapFunction<Row, Row>) row -> RowFactory.create(row.getInt(0), "spark_" + row.getString(1), row.getDate(2)), RowEncoder.apply(rowDataset.schema()));
54. //显示数据
55. //dataset.show();
56. Properties properties = new Properties();
57. properties.put("user", destUserName);
58. properties.put("password", destPassword);;
59. dataset.write()
60. .mode(SaveMode.Append)
61. .jdbc("jdbc:mysql://" + destHost + ":" + destPort + "/" + destDataBase,destTable, properties);
62. }
64. }
在上述代码中，执行了简单的 ETL 操作（筛选 id > 2 的数据，并在 name 字段前添加前缀 “spark_“），并将处理后的数据写入到 MatrixOne 数据库中。
步骤四：查看执行结果
在 MatrixOne 中执行如下 SQL 查看执行结果：
1. select * from test.person;
2. +------+---------------+------------+
3. | id | name | birthday |
4. +------+---------------+------------+
5. | 3 | spark_wangwu | 2023-07-12 |
6. | 4 | spark_zhaoliu | 2023-08-07 |
7. +------+---------------+------------+
8. 2 rows in set (0.01 sec)
示例 2：将 Hive 数据导入到 MatrixOne
步骤一：初始化项目
-
启动 IDEA，点击 File > New > Project，选择 Spring Initializer，并填写以下配置参数：
-
Name：mo-spark-demo
- Location：~\Desktop
- Language：Java
- Type：Maven
- Group：com.example
- Artiface：matrixone-spark-demo
- Package name：com.matrixone.demo
- JDK 1.8
2. 添加项目依赖，在项目根目录下的 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>com.example.mo</groupId>
8. <artifactId>mo-spark-demo</artifactId>
9. <version>1.0-SNAPSHOT</version>
11. <properties>
12. <maven.compiler.source>8</maven.compiler.source>
13. <maven.compiler.target>8</maven.compiler.target>
14. <spark.version>3.2.1</spark.version>
15. </properties>
17. <dependencies>
18. <dependency>
19. <groupId>org.apache.spark</groupId>
20. <artifactId>spark-sql_2.12</artifactId>
21. <version>${spark.version}</version>
22. </dependency>
24. <dependency>
25. <groupId>org.apache.spark</groupId>
26. <artifactId>spark-hive_2.12</artifactId>
27. <version>${spark.version}</version>
28. </dependency>
29. <dependency>
30. <groupId>org.apache.spark</groupId>
31. <artifactId>spark-catalyst_2.12</artifactId>
32. <version>${spark.version}</version>
33. </dependency>
34. <dependency>
35. <groupId>org.apache.spark</groupId>
36. <artifactId>spark-core_2.12</artifactId>
37. <version>${spark.version}</version>
38. </dependency>
39. <dependency>
40. <groupId>org.codehaus.jackson</groupId>
41. <artifactId>jackson-core-asl</artifactId>
42. <version>1.9.13</version>
43. </dependency>
44. <dependency>
45. <groupId>org.codehaus.jackson</groupId>
46. <artifactId>jackson-mapper-asl</artifactId>
47. <version>1.9.13</version>
48. </dependency>
51. <dependency>
52. <groupId>mysql</groupId>
53. <artifactId>mysql-connector-java</artifactId>
54. <version>8.0.16</version>
55. </dependency>
57. </dependencies>
59. </project>
步骤二：准备 Hive 数据
在终端窗口中执行以下命令，创建 Hive 数据库、数据表，并插入数据：
1. hive
2. hive> create database motest;
3. hive> CREATE TABLE `users`(
4. `id` int,
5. `name` varchar(255),
6. `age` int);
7. hive> INSERT INTO motest.users (id, name, age) VALUES(1, 'zhangsan', 12),(2, 'lisi', 17),(3, 'wangwu', 19);
步骤三：创建 MatrixOne 数据表
在 node3 上，使用 MySQL 客户端连接到 node1 的 MatrixOne。然后继续使用之前创建的 “test” 数据库，并创建新的数据表 “users”。
1. CREATE TABLE `users` (
2. `id` INT DEFAULT NULL,
3. `name` VARCHAR(255) DEFAULT NULL,
4. `age` INT DEFAULT NULL
5. )
步骤四：拷贝配置文件
将 Hadoop 根目录下的 “etc/hadoop/core-site.xml” 和 “hdfs-site.xml” 以及 Hive 根目录下的 “conf/hive-site.xml” 这三个配置文件复制到项目的 “resource” 目录中。
步骤五：编写代码
在 IntelliJ IDEA 中创建名为 “Hive2Mo.java” 的类，用于使用 Spark 从 Hive 中读取数据并将数据写入 MatrixOne。
1. package com.matrixone.spark;
3. import org.apache.spark.sql.*;
5. import java.sql.SQLException;
6. import java.util.Properties;
8. /**
9. * @auther MatrixOne
10. * @date 2022/2/9 10:02
11. * @desc
12. *
13. * 1.在 hive 和 matrixone 中分别创建相应的表
14. * 2.将 core-site.xml hdfs-site.xml 和 hive-site.xml 拷贝到 resources 目录下
15. * 3.需要设置域名映射
16. */
17. public class Hive2Mo {
19. // parameters
20. private static String master = "local[2]";
21. private static String appName = "app_spark_demo";
23. private static String destHost = "192.168.146.10";
24. private static Integer destPort = 6001;
25. private static String destUserName = "root";
26. private static String destPassword = "111";
27. private static String destDataBase = "test";
28. private static String destTable = "users";
31. public static void main(String[] args) throws SQLException {
32. SparkSession sparkSession = SparkSession.builder()
33. .appName(appName)
34. .master(master)
35. .enableHiveSupport()
36. .getOrCreate();
38. //SparkJdbc 读取表内容
39. System.out.println("读取 hive 中 person 的表内容");
40. // 读取表中所有数据
41. Dataset<Row> rowDataset = sparkSession.sql("select * from motest.users");
42. //显示数据
43. //rowDataset.show();
44. Properties properties = new Properties();
45. properties.put("user", destUserName);
46. properties.put("password", destPassword);;
47. rowDataset.write()
48. .mode(SaveMode.Append)
49. .jdbc("jdbc:mysql://" + destHost + ":" + destPort + "/" + destDataBase,destTable, properties);
50. }
52. }
步骤六：查看执行结果
在 MatrixOne 中执行如下 SQL 查看执行结果：
1. mysql> select * from test.users;
2. +------+----------+------+
3. | id | name | age |
4. +------+----------+------+
5. | 1 | zhangsan | 12 |
6. | 2 | lisi | 17 |
7. | 3 | wangwu | 19 |
8. +------+----------+------+
9. 3 rows in set (0.00 sec)
