使用 Apache SeaTunnel 在 MySQL 和 HTTP 之间的数据同步示例

🕗 发布于 2024-10-11 05:55 大数据

随着现代企业数据量的不断增长，跨系统、跨平台的数据同步需求变得愈发重要。

在实际的业务场景中，开发者常常需要将数据从 MySQL 同步到其他系统，或者从不同的数据源同步回 MySQL。Apache SeaTunnel 作为一款高效的分布式数据集成平台，支持批处理和流处理，能够灵活地完成这些任务。

本文将详细介绍如何使用 Apache SeaTunnel 实现以下几种常见的数据同步场景：

MySQL 同步到 HTTP 接口
MySQL 同步到 MySQL
HTTP 接口同步到 MySQL
MySQL-CDC 同步到 HTTP 接口

我们将逐一展示这些同步场景的配置方式，并提供清晰的代码示例，帮助开发者快速掌握 SeaTunnel 在不同场景下的应用。

官方文档参考：

SeaTunnel JDBC Source Connector

前置准备

在开始之前，请确保已经下载了对应版本的 MySQL JDBC 驱动 mysql-connector-java-xxx.jar，并将其放置在 SeaTunnel 的安装目录下的 lib 文件夹中。

可以从以下链接获取：https://mvnrepository.com/artifact/mysql/mysql-connector-java

对于使用 Spark 或 Flink 的 SeaTunnel 任务，也需要将该 JAR 包复制到相应的目录下：

Spark: $SPARK_HOME/jars/
Flink: $FLINK_HOME/lib/

接下来，我们将逐一展示四种数据同步的配置和代码示例。

MySQL 同步到 HTTP 接口

在此场景中，我们将 MySQL 数据表中的信息同步到指定的 HTTP 接口。

这里假设我们从 user_info 表中查询数据并通过 HTTP POST 请求将其发送到目标 API。

env {
  execution.parallelism = 2
  job.mode = "BATCH"  # MySQL 作为数据源，只支持批量同步
}

source {
   jdbc {
     url =  "jdbc:mysql://172.27.10.22:6033/test"
     driver = "com.mysql.cj.jdbc.Driver"
     connection_check_timeout_sec = 100
     user = "root"
     password = "root"
     query = "SELECT * FROM user_info ORDER BY create_time LIMIT 1"
     result_table_name = "user_info_out"
  }
}

transform {
    Sql {
      source_table_name = "user_info_out"
      result_table_name = "user_info_sink"
      query = "select info, user_name, age from user_info_out"
    }
}

sink {
  Console {
    source_table_name = "user_info_sink"
  }

  http {
    source_table_name = "user_info_sink"
    url = "https://test.test.com:8080/api/user/test"
    method = "POST"
    headers = {Accept="application/json", Content-Type="application/json;charset=utf-8"}
  }
}

MySQL 同步到 MySQL

在此示例中，我们将从一个 MySQL 数据库中提取数据，并将其同步到另一个 MySQL 数据库。此场景适用于多个数据库实例之间的数据迁移或备份。

env {
  execution.parallelism = 2
  job.mode = "BATCH"
}

source {
    Jdbc {
        url =  "jdbc:mysql://172.27.10.22:6033/test"
        driver = "com.mysql.cj.jdbc.Driver"
        connection_check_timeout_sec = 10
        user = "root"
        password = "root"
        query = "SELECT `name`,`score` FROM `user`"
        result_table_name = "user_info"
    }
}

sink {
  Jdbc {
        source_table_name = "user_info"
        url =  "jdbc:mysql://192.27.10.22:16033/temp_user"
        driver = "com.mysql.cj.jdbc.Driver"
        connection_check_timeout_sec = 100
        user = "root"
        password = "root"
        query = "INSERT INTO `student`(`name`, `score`) VALUES(?, ?)"
  }
}

HTTP 接口同步到 MySQL

本示例展示了如何将 HTTP 接口中的数据同步到 MySQL 数据库。

这在从第三方 API 获取数据并将其存储到本地数据库的场景中非常实用。

env {
  execution.parallelism = 2
  job.mode = "STREAMING"  # HTTP 作为数据源，支持批量和流式模式
  checkpoint.interval = 10000  # 执行间隔（毫秒）
}

source {
  Http {
    url = "https://test.test.com:8080/api/test"
    method = "GET"
    format = "json"
    headers = {Authorization="Bearer example-token", language="zh"}
    params = {userId="fa438165b2c84d8dbe9175d152718437"}
    content_field = "$.content.*"
    schema = {
      fields {
        userId = string
        age = int
        phone = string
        name = string
      }
    }
    result_table_name = "user_info"
  }
}

transform {
    Sql {
      source_table_name = "user_info"
      result_table_name = "user_info_out"
      query = "SELECT name as userName, userId, age, phone FROM user_info"
    }
}

sink {
  Jdbc {
     url = "jdbc:mysql://172.27.10.22:26033/test"
     driver = "com.mysql.cj.jdbc.Driver"
     connection_check_timeout_sec = 100
     user = "root"
     password = "root"
     source_table_name = "user_info_out"
     query = "INSERT INTO `user_bak`(`userName`, `userId`, `age`, `phone`) VALUES (?, ?, ?, ?)"
  }
}

MySQL-CDC 同步到 HTTP 接口

MySQL-CDC（Change Data Capture）允许实时捕获数据库中的数据变化。

在此示例中，我们将 MySQL 数据库中的变化通过 CDC 机制捕获，并将其同步到 HTTP 接口。

env {
  execution.parallelism = 2
  job.mode = "STREAMING"  # MySQL-CDC 支持批量和流式模式
  checkpoint.interval = 10000  # 执行间隔（毫秒）
}

source {
    MySQL-CDC {
      catalog = {
        factory = MySQL
      }
      base-url = "jdbc:mysql://${mysql_ip_port}/test?serverTimezone=Asia/Shanghai&characterEncoding=utf8&useSSL=false"
      username = ${mysql_username}  # 使用变量替换
      password = ${mysql_pass}  # 使用变量替换
      table-names = ["test.user"]
      startup.mode = "initial"
      result_table_name = "user_info_out"
      table-names-config = [
        {
          table = "test.user"
          primaryKeys = ["user_id"]
        }
      ]
    }
}

transform {
    FilterRowKind {
      source_table_name = "user_info_out"
      result_table_name = "user_info_sink"
      include_kinds = ["UPDATE_AFTER", "INSERT"]
    }
}

sink {
  http {
    source_table_name = "user_info_sink"
    url = "https://test.test.com:28080/api/user/test"
    method = "POST"
    headers = {Accept="application/json", Content-Type="application/json;charset=utf-8"}
  }
}

总结

通过 Apache SeaTunnel 的强大数据集成能力，开发者可以轻松实现多种数据源之间的同步操作。无论是数据库与 API 之间的数据传输，还是跨数据库的数据迁移，SeaTunnel 都为开发者提供了灵活、高效的解决方案。

希望通过本文的示例，您能够快速上手并在实际项目中 应用 SeaTunnel 进行复杂的数据同步任务。

SeaTunnel 提供的流处理和批处理模式极大地满足了多种场景下的数据处理需求，使得跨平台、跨数据源的数据集成变得更加简单、高效。

本文由白鲸开源科技提供发布支持！

原文地址：https://blog.csdn.net/weixin_54625990/article/details/142815676

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：大数据-161 Apache Kylin 构建Cube 按照日期、区域、产品、渠道与 Cube 优化
下一篇：嵌入式硬件设计详解

股市入门常见术语介绍
股市入门常见术语科普
阅读更多2024-10-11
金融期货市场中的对冲策略及应用
对冲策略通过期货合约的买卖，帮助市场参与者减轻价格波动的影响，从而在不确定的市场环境中获得稳定的收益。例如，某一商品期货在纽约交易所和伦敦交易所的价格存在差异时，投资者可以在价格较低的市场买入，同时在
阅读更多2024-10-11
网优学习干货：王者荣耀游戏用户体验洞察及质差识别（1）
针对热点游戏（王者荣耀）进行业务质量评估，并通过对端到端定界分析，从无线、核心网、互联网维度识别影响用户体验关键因素，为游戏用户的体验优化提供依据。
阅读更多2024-10-11
静态IP与动态IP的选择分析
静态IP与动态IP的选择分析
阅读更多2024-10-11
股指期货的杠杆是怎么体现和使用的？
如果账户中的资金低于维持保证金所要求的金额，投资者将收到一个保证金催缴通知，要求向账户中注入更多资金，否则账户将被强制平仓。股指期货的杠杆效应是通过保证金交易实现的。例如，如果一个股指期货合约的价值为
阅读更多2024-10-11
基于STM32 ARM+FPGA+AD的电能质量分析仪方案设计(一）硬件设计
以确保其具有足够的计算能力和运行速度处理复杂的运算，从而保证系统的效率、和维护性更高，需要大量人力和物力投入，增加了系统的开发和维护成本。高效的计算和处理任务。处理器，提供更完善的功能和更强大的性能，
阅读更多2024-10-11
高级java每日一道面试题-2024年10月9日-JVM篇-什么是双亲委派模型?
双亲委派模型（Parent Delegation Model）是Java类加载器（ClassLoader）的一种工作模式。这种模型确保了Java类的加载过程具有良好的层次结构和安全性，避免了类的重复加
阅读更多2024-10-11
[linux] 在VMware中安装linux、文件下载及详细安装过程（附下载链接）
链接：https://pan.quark.cn/s/b99760438175。选择安装程序光盘映像文件，点击浏览，找到刚才下载的.iso文件，点击下一步。设置镜像最大大小，默认20即可（实际会使用3G
阅读更多2024-10-11
Django-rest-framework(DRF)怎么实现Excel文件导出
a、创建工作簿openpyxl.Workbook() #创建一个新的工作簿对象。
阅读更多2024-10-11
markdown里粘贴图片的同时保存路径在vscode里实现
下载扩展Markdown Image。
阅读更多2024-10-11