使用Apache Camel从AWS S3读取CSV文件教程及日志排障指南

import org.apache.camel.main.Main;
import org.apache.camel.impl.DefaultCamelContext;

public class S3FileReaderApplication {

  public static void main(String[] args) throws Exception {
    // 创建一个Camel上下文
    var camelContext = new DefaultCamelContext();

    // 添加自定义路由
    camelContext.addRoutes(new S3FileProcessingRoute());

    // 启动Camel上下文
    System.out.println("Starting Camel context...");
    camelContext.start();

    // 让应用运行一段时间，以便路由可以处理文件
    // 实际应用中，这通常是一个持续运行的服务
    Thread.sleep(10_000); // 运行10秒

    // 停止Camel上下文
    System.out.println("Stopping Camel context...");
    camelContext.stop();
    System.out.println("Camel context stopped.");
  }
}

import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.model.dataformat.CsvDataFormat;

public class S3FileProcessingRoute extends RouteBuilder {

  @Override
  public void configure() {
    // S3组件URI配置
    // mybucket.com: 替换为您的S3存储桶名称
    // prefix=etl/hello.csv: 指定要读取的文件前缀。这里是读取特定文件。
    // useDefaultCredentialsProvider=true: 使用默认的AWS凭证提供者链（例如环境变量、IAM角色、配置文件）。
    // deleteAfterRead=false: 读取后不删除S3对象。
    // maxMessagesPerPoll=1: 每次轮询只处理一个S3对象。
    var s3Url = String.format(
        "aws2-s3://mybucket.com?"
            + "prefix=etl/hello.csv&useDefaultCredentialsProvider=true&deleteAfterRead=false&maxMessagesPerPoll=1");

    System.out.println("Route configuration started...");

    // 定义路由：从S3读取文件，解组为CSV，然后记录到日志
    from(s3Url)
        .routeId("S3CsvFileReader") // 为路由设置一个ID
        .unmarshal(new CsvDataFormat()) // 将S3对象内容解组为CSV格式
        .split(body()) // 将CSV行分割成单独的消息
        .log("Processing CSV record: ${body}") // 记录每条CSV记录
        .end();

    System.out.println("Route configuration finished.");
  }
}

<properties>
    <log4j2.version>2.20.0</log4j2.version> <!-- 根据您的需求选择合适的版本 -->
    <camel.version>3.19.0</camel.version> <!-- 确保与您的Camel版本匹配 -->
    <java.version>17</java.version>
</properties>

<dependencies>
    <!-- Apache Camel Core -->
    <dependency>
        <groupId>org.apache.camel</groupId>
        <artifactId>camel-core</artifactId>
        <version>${camel.version}</version>
    </dependency>
    <!-- Apache Camel AWS2 S3 Component -->
    <dependency>
        <groupId>org.apache.camel</groupId>
        <artifactId>camel-aws2-s3</artifactId>
        <version>${camel.version}</version>
    </dependency>
    <!-- Apache Camel CSV DataFormat -->
    <dependency>
        <groupId>org.apache.camel</groupId>
        <artifactId>camel-csv</artifactId>
        <version>${camel.version}</version>
    </dependency>

    <!-- Log4j2 Logging Dependencies -->
    <dependency>
        <groupId>org.apache.logging.log4j</groupId>
        <artifactId>log4j-api</artifactId>
        <version>${log4j2.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.logging.log4j</groupId>
        <artifactId>log4j-core</artifactId>
        <version>${log4j2.version}</version>
    </dependency>
    <!-- SLF4J to Log4j2 binding -->
    <dependency>
        <groupId>org.apache.logging.log4j</groupId>
        <artifactId>log4j-slf4j-impl</artifactId>
        <version>${log4j2.version}</version>
    </dependency>
    <!-- 可选：如果您的Camel版本是3.x，通常需要这个以避免冲突 -->
    <dependency>
        <groupId>org.apache.logging.log4j</groupId>
        <artifactId>log4j-to-slf4j</artifactId>
        <version>${log4j2.version}</version>
    </dependency>
</dependencies>