使用 Spring Batch 实现 ETL

Posted 2023-02-27

【中文标题】使用 Spring Batch 实现 ETL

【英文标题】：ETL implementation using Spring Batch

【发布时间】：2016-01-02 19:36:06

【问题描述】：

我需要为正在进行的其中一个项目实施 ETL 应用程序。

它有以下步骤：

需要从表中读取一些值 作为 Job 参数传入。 第 1 步返回的对象将进一步用于检索 来自第二个表的一些数据。 然后必须从一个平面文件中读取，该文件将与 步骤 2 中的值。应用业务逻辑。然后写入表格。

我们正在使用 Spring Data JPA，Spring 集成。

我面临的挑战是从表中读取值以检索作业的参数，然后启动作业。

然后步骤 2 的输出必须与文件信息一起发送以进行进一步处理。

我知道如何独立实施上述步骤，但很难将它们从头到尾联系起来。

分享设计上述内容的任何想法都会很棒。提前致谢。

【参考方案1】：

我会尝试为您的不同点提供一些想法。

1 - 读取表值并将它们作为作业参数传递

我在这里看到 2 个解决方案：

您可以执行“手动”查询（即不使用 springbatch），然后执行业务逻辑以将结果作为 JobParameters 传递（您只需要 JobLauncher 或 CommandLineJobRunner，请参阅 Springbatch Documentation §4.4）：

JobLauncher jobLauncher = (JobLauncher) context.getBean("jobLauncher");
Job job = (Job) context.getBean(jobName);

// Do your business logic and your database query here.

// Create your parameters
JobParameter parameter = new JobParameter(resultOfQuery);

// Add them to a map
Map<String, JobParameter> parameters = new HashMap<String, JobParameter>();
parameters.add("yourParameter", parameter);

// Pass them to the job
JobParameters jobParameters = new JobParameters(parameters);
JobExecution execution = jobLauncher.run(job, parameters);

另一种解决方案是添加JobExecutionListener 并覆盖方法beforeJob 进行查询，然后将结果保存在executionContext 中（然后您可以使用：#jobExecutionContext[name] 访问）。

@Override
public void beforeJob(JobExecution jobExecution) 

    // Do your business logic and your database query here.

    jobExecution.getExecutionContext().put(key, value);

在每种情况下，您都可以使用 SpringBatch ItemReader 进行查询。例如，您可以将 item reader 声明为侦听器的字段（不要忘记 setter）并将其配置为：

<batch:listener>
    <bean class="xx.xx.xx.YourListener">
        <property name="reader">
            <bean class="org.springframework.batch.item.database.JdbcCursorItemReader">
                <property name="dataSource" ref="dataSource"></property>
                <property name="sql" value="$yourSQL"></property>
                 <property name="rowMapper">
                    <bean class="xx.xx.xx.YourRowMapper"></bean>
                </property>
            </bean>
        </property>
    </bean>
</batch:listener>

2 - 根据上一步的结果读取表格

再一次，您可以使用JobExecutionContext 在步骤之间存储和检索数据。然后，您可以实现 StepExecutionListener 来覆盖方法 beforeStep 并访问 StepExecution 这将引导您到 JobExecution。

3 - 将表格读取结果与文件读取结果一起发送

没有“默认”CompositeItemReader 可以让您同时阅读 2 个来源，但我认为这不是您真正想要做的。

对于您的情况，我将在 <batch:chunk> 中将“表格阅读器”声明为阅读器，然后声明一个自定义 ItemProcessor，它将具有另一个 ItemReader 字段。这位读者将是您的FlatFileItemReader。然后，您可以手动开始读取并在 process 方法中应用您的业务逻辑。

【讨论】：

谢谢亚历克斯。让我试试建议的方法，看看效果如何