使用 Spark 将批量数据写入 MatrixOne

概述

Apache Spark 是一个为高效处理大规模数据而设计的分布式计算引擎。它采用分布式并行计算的方式，将数据拆分、计算、合并的任务分散到多台计算机上，从而实现了高效的数据处理和分析。

应用场景

  • 大规模数据处理与分析

    Spark 能够处理海量数据，通过并行计算任务提高了处理效率。它广泛应用于金融、电信、医疗等领域的数据处理和分析。

  • 流式数据处理

    Spark Streaming 允许实时处理数据流，将其转化为可供分析和存储的批处理数据。这在在线广告、网络安全等实时数据分析场景中非常有用。

  • 机器学习

    Spark 提供了机器学习库（MLlib），支持多种机器学习算法和模型训练，用于推荐系统、图像识别等机器学习应用。

  • 图计算

    Spark 的图计算库（GraphX）支持多种图计算算法，适用于社交网络分析、推荐系统等图分析场景。

本篇文档将介绍两种使用 Spark 计算引擎实现批量数据写入 MatrixOne 的示例。一种示例是从 MySQL 迁移数据至 MatrixOne，另一种是将 Hive 数据写入 MatrixOne。

前期准备

硬件环境

本次实践对于机器的硬件要求如下：

服务器名称服务器 IP安装软件操作系统
node1192.168.146.10MatrixOneDebian11.1 x86
node3192.168.146.11IDEA、MYSQL、Hadoop、HiveWindows 10

软件环境

本次实践需要安装部署以下软件环境：

示例 1：从 MySQL 迁移数据至 MatrixOne

步骤一：初始化项目

  1. 启动 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 后，创建演示所需的数据库以及数据表。

  1. 在 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;
    3. import org.apache.spark.sql.Dataset;
    4. import org.apache.spark.sql.Row;
    5. import org.apache.spark.sql.SQLContext;
    6. import org.apache.spark.sql.SparkSession;
    8. import java.util.Properties;
    10. /**
    11.  * @auther MatrixOne
    12.  * @desc 读取 MatrixOne 数据
    13.  */
    14. public class MoRead {
    16.     // parameters
    17.     private static String master = "local[2]";
    18.     private static String appName = "mo_spark_demo";
    20.     private static String srcHost = "192.168.146.10";
    21.     private static Integer srcPort = 6001;
    22.     private static String srcUserName = "root";
    23.     private static String srcPassword = "111";
    24.     private static String srcDataBase = "test";
    25.     private static String srcTable = "person";
    27.     public static void main(String[] args) {
    28.         SparkSession sparkSession = SparkSession.builder().appName(appName).master(master).getOrCreate();
    29.         SQLContext sqlContext = new SQLContext(sparkSession);
    30.         Properties properties = new Properties();
    31.         properties.put("user", srcUserName);
    32.         properties.put("password", srcPassword);
    33.         Dataset<Row> dataset = sqlContext.read()
    34.                 .jdbc("jdbc:mysql://" + srcHost + ":" + srcPort + "/" + srcDataBase,srcTable, properties);
    35.         dataset.show();
    36.     }
    38. }

  3. 在 IDEA 中运行 MoRead.Main()，执行结果如下：

    使用 Spark 将批量数据写入 MatrixOne - 图2

步骤三：将 MySQL 数据写入 MatrixOne

现在可以开始使用 Spark 将 MySQL 数据迁移到 MatrixOne。

  1. 准备 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

步骤一：初始化项目

  1. 启动 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)