seatunnel_了解及使用

2024-05-19 本文已影响0人 Yluozi

seatunnel connector 连接器

source 连接器：读取原数据的连接器，支持数据源种类包括：关系型，非关系型，图形数据库等等57种数据库。

transform 连接器：转换器---将原数据源数据转换成目标源连接器的工具。

slink 连接器：写入目标数据源连接器，支持关系型，非关系型，图形数据库等等55种数据库

image.png

执行引擎选择

SeaTunnel Engine：
这是SeaTunnel自带的执行引擎，用于数据同步。它提供了高效且稳定的数据处理能力。
Apache Spark：
Spark是一个快速、通用的大规模数据处理引擎。SeaTunnel可以与Spark紧密集成，利用Spark的分布式计算能力来处理数据流。
Apache Flink：
Flink是一个流处理框架，用于无边界和有边界数据流的处理。SeaTunnel支持Flink引擎，允许用户构建实时数据流处理应用程序。

部署

（略）
注意多节点情况下本地文件读取时需要做策略处理，不然会存在部分时候报找不到文件错误。

查看部署引擎

# 假设你已经将 SeaTunnel 的 bin 目录添加到了 PATH 环境变量中  
# 你可以使用 --help 或 -h 参数来获取帮助信息  
  
# 对于基于 Spark 的 SeaTunnel  
./start-waterdrop-spark.sh --help  
  
# 或者对于其他可能的启动脚本  
./start-seatunnel.sh --help  
  
# 或者查看某个具体配置文件的帮助（如果支持的话）  
./start-waterdrop-spark.sh -c config/your-config.conf --help

执行文件样例

env {                                                                                                                                                                          
  # You can set SeaTunnel environment configuration here                                                                                                                      
  execution.parallelism = 10                                                                                                                                                  
  job.mode = "BATCH"                                                                                                                                                           
  checkpoint.interval = 10000                                                                                                                                                 
}                                                                                                                                                                              
                                                                                                                                                                    
source {                                                                                                                                                                                                                       
jdbc {
                result_table_name = "fake1"
        url = "jdbc:oracle:thin:@10.1.0.10:1521:ora11g"
        driver = "oracle.jdbc.OracleDriver"
        user = "name"
        password = "pwd"
        batchSize = 1000
        query = "select * from  ALARM_LOG"
    }              
}                                                                                                                                                                               

transform {  
}  

sink {
  Console {
      #控制台打印表数据
      source_table_name = "fake1"
  }
}

四部分：

env：在SeaTunnel中，env通常用于设置运行环境的相关参数或配置。

source：用于定义SeaTunnel需要在何处获取数据，并使用获取的数据进行下一步操作。你可以同时定义多个源，并且每个源都有自己的特定参数来定义如何获取数据。SeaTunnel会提取每个源将使用的参数，如result_table_name参数，用于指定当前源生成的数据名称，方便其他模块后续使用

transform：当有了数据源后，业务可能需要进一步处理数据，因此SeaTunnel提供了转换模块transform。在SeaTunnel中，transform只能用于对数据进行一些简单的转换，例如将一列的数据转换为大写或小写，更改列名，或者将一列拆分为多列。同时，transform作为一个中间处理模块，可以同时使用result_table_name和source_table_name进行配置。

sink:sink用于定义SeaTunnel需要把数据写到哪里。当定义多个源、转换时，每个源和转换读取哪些数据，需要使用result_table_name和source_table_name两个键进行配置。每个源端将配置一个result_table_name，以指示生成的数据源的名称，其他转换和接收模块可以使用source_table_name来引用对应的数据源名称。

其中源数据写法：

jdbc连接本地文件:

LocalFile{
    result_table_name = "fake1"
    delimiter = "\\|"
    ignore_first_line = false
    path = "/tmp/seatunnel/alarmlog.txt"
    file_format_type = "text"
    datatime_format = "yyyy-MM-dd HH:mm:ss"
    schema {
      fields {
          ID = string
          MSG= string
      }
    }
  }

jdbc连接oracle:

 jdbc {
        url = "jdbc:oracle:thin:@10.1.0.10:1521:ora11g"
        driver = "oracle.jdbc.OracleDriver"
        user = "name"
        password = "pwd"
        batchSize = 1000
        query = "select * from  ALARM_LOG"
    }              
}

jdbc连接mysql:

  jdbc {
        url = "jdbc:mysql://10.1.1.1:3306/demo?useSSL=false&useUnicode=true&characterEncoding=utf-8"
        driver = "com.mysql.cj.jdbc.Driver"
        user = "root"
        password = "root"
        batchSize = 1000
        result_table_name = "fake1"
        query = "select * from d_sesl"
    }

其中目标数据写法（sink）：

#打印控制台
  Console {
  }
#下载到数据库
jdbc {
        url = "jdbc:mysql://10.1.1.10:3306/demo?useSSL=false&useUnicode=true&characterEncoding=utf-8"
        driver = "com.mysql.cj.jdbc.Driver"
        user = "root"
        password = "1234"
        batchSize = 1000
        writeMode = "insert" # 或者使用 "update" 根据需要
        fieldNames = ["id","je","se"] # 对应目标表的字段名           
        query = "insert into d_sesl(id,je,se) values(?,?,?)"
    }

启动执行文件

查看seatunnel执行情况

#ps aux 和ps -ef
 ps aux | grep seatunnel

启动执行文件并查看结果

# bin 目录下，命令
#全日志 执行本地节点？
./bin/seatunnel.sh --config <your-config-file.conf>  -e  local
#简日志
./bin/seatunnel.sh --config <your-config-file.conf>