【Demo】使用 SeaTunnel 将数据从 Kafka 迁移到 Redis

mvn clean package -pl seatunnel-dist -am -Dmaven.test.skip=true
复制

mvn clean package -pl seatunnel-examples/seatunnel-engine-examples -am -Dmaven.test.skip=true -T 1C
复制

public class SeaTunnelEngineExample {


    public static void main(String[] args)
            throws FileNotFoundException, URISyntaxException, CommandException {
        String configurePath = args.length > 0 ? args[0] : "/examples/fake_to_console.conf";
        String configFile = getTestConfigFile(configurePath);
        ClientCommandArgs clientCommandArgs = new ClientCommandArgs();
        clientCommandArgs.setConfigFile(configFile);
        clientCommandArgs.setCheckConfig(false);
        clientCommandArgs.setJobName(Paths.get(configFile).getFileName().toString());
         Change Execution Mode to CLUSTER to use client mode, before do this, you should start
         SeaTunnelEngineServerExample
        clientCommandArgs.setMasterType(MasterType.LOCAL);
        SeaTunnel.run(clientCommandArgs.buildCommand());
    }
复制

env {
    execution.parallelism = 2
    job.mode = "BATCH"
  }
source {
    Jdbc {
      url = "jdbc:mysql://xxxxxx:3306/xxxxx"
      driver = "com.mysql.cj.jdbc.Driver"
      connection_check_timeout_sec = 100
      user = "root"
      password = "xxxxx"
      query = "select * from test"
      #partition_column = "id"
      #partition_num = 10
    }
  }
  sink {
    Redis {
      host = xxxxxx
      port = 6379
      key = "seatunnel_jdbc"
      data_type = list
      auth = "xxxxxx"
    }
  }
复制

java.lang.RuntimeException: Plugin PluginIdentifier{engineType='seatunnel', pluginType='source', pluginName='XXXX'} not found.
复制

<dependency>
<groupId>org.apache.seatunnel</groupId>
<artifactId>connector-rabbitmq</artifactId>
<version>${project.version}</version>
</dependency>
复制

mvn clean package -pl seatunnel-dist -am -Dmaven.test.skip=true
复制