标签: spring-batch

我有一个使用Spring Batch和Spring MVC的应用程序.我能够将Spring Batch Admin作为一个单独的战争部署,并将其用于我的应用程序使用的同一个DB,尽管我想将它集成到我自己的应用程序中,也可能修改一些视图.

有没有一种简单的方法可以做到这一点,还是我必须分叉并从那里开始？

作为Java开发人员,我习惯使用Spring Batch进行批处理,通常使用流式库来导出大型XML文件,例如StAX.

我现在正在开发Scala应用程序,并想知道是否有任何框架,工具或指南来实现批处理.

我的Scala应用程序使用了Cake Pattern,我不知道如何将它与SpringBatch集成.此外,我想遵循所描述的指导方针Functional programming in Scala并尝试保持功能纯度,使用像IO monad这样的东西......

我知道这是一个悬而未决的问题,但我从未读过任何有关此事的内容......

有人已经在这里实现了功能批处理吗？它是如何工作的？我应该有一个main在IO monad中创建一个批处理操作并运行它吗？是否有任何工具或指南可以帮助,监控或处理可重启性,就像我们在Java中使用Spring Batch一样.你在Scala中使用Spring Batch吗？如何处理集成部分,例如等待JMS/AMQP消息开始生成XML的处理？

欢迎任何关于子喷气机的反馈

根据文档, spring批处理管理员很容易嵌入到现有的应用程序中.只需复制web.xml和index.jsp,然后添加所需的依赖项就足以让它工作.

但是,如果我想在现有的春季启动项目中使用它,它会变得更糟.根据这个例子,配置有点hacky但它​​的工作原理.我尝试@EnableBatchProcessing在配置bean中使用注释.然后我得到以下异常.

Exception in thread "main" org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'jobBuilders' defined in class path resource [org/springframework/batch/core/configuration/annotation/SimpleBatchConfiguration.class]: Instantiation of bean failed; nested exception is org.springframework.beans.factory.BeanDefinitionStoreException: Factory method [public org.springframework.batch.core.configuration.annotation.JobBuilderFactory org.springframework.batch.core.configuration.annotation.AbstractBatchConfiguration.jobBuilders() throws java.lang.Exception] threw exception; nested exception is java.lang.ClassCastException: org.springframework.batch.core.repository.support.JobRepositoryFactoryBean$$EnhancerBySpringCGLIB$$49fa0273 cannot be cast to org.springframework.batch.core.repository.JobRepository
    at org.springframework.beans.factory.support.ConstructorResolver.instantiateUsingFactoryMethod(ConstructorResolver.java:597)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.instantiateUsingFactoryMethod(AbstractAutowireCapableBeanFactory.java:1095)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBeanInstance(AbstractAutowireCapableBeanFactory.java:990)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:504)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:475)
    at org.springframework.beans.factory.support.AbstractBeanFactory$1.getObject(AbstractBeanFactory.java:302)
    at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.getSingleton(DefaultSingletonBeanRegistry.java:228)
    at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:298)
    at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:193)
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:706)
    at org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:762)
    at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:482)
    at org.springframework.boot.context.embedded.EmbeddedWebApplicationContext.refresh(EmbeddedWebApplicationContext.java:109)
    at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:691) …

我正在进行独立弹簧批量作业.在JUNIT中,这完全正常

@RunWith(SpringJUnit4ClassRunner.class)
//@SpringApplicationConfiguration(classes = KPBootApplication.class)
@ContextConfiguration(locations={"classpath:kp-sb.xml"})
public class KPBootApplicationTests {

    private final static Logger LOG=LoggerFactory.getLogger(KPBootApplicationTests.class);

    @Autowired
    ApplicationContext context;

    @Autowired
    private JobLauncher jobLauncher;

    @Autowired
    private Job job;

    @Test
    public void testJob() {

        final JobParameters jobParameters = new JobParametersBuilder()
                .toJobParameters();

        JobExecution execution;
        try {
            execution = jobLauncher.run(job, jobParameters);
            final ExitStatus status = execution.getExitStatus();

            if (ExitStatus.COMPLETED.getExitCode().equals(status.getExitCode())) {
                LOG.info("Job completed Yeaaaaaaaa!!!!!");
            } else {
                final List<Throwable> exceptions = execution
                        .getAllFailureExceptions();
                for (final Throwable throwable : exceptions) {
                    LOG.error(throwable.getMessage(), throwable);

                }
            }
        } catch (JobExecutionAlreadyRunningException …

我们有简单的批处理工作,工作正常.最近,我们有新的需求来实现新的批处理以生成报告.我们有差异的数据源来阅读以准备此报告.具体来说,每个报告可能有一个视图.

现在我们希望以这样一种方式扩展这个过程,它可以扩展并尽早完成.

我熟悉多线程步骤但不确定其他策略(远程分块和分区步骤)以及何时使用.

在我们的案例中,处理+写入文件是更多的资源激励然后阅读.

在这种情况下哪种方法最适合.

或者,如果我们发现db中的读取数据与写入+处理到文件的资源激励相同,那么我们必须改进/扩展此过程的最佳选择是什么.

我有一个以下的Spring Batch Job配置:

@Configuration
@EnableBatchProcessing
public class JobConfig {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job job() {
        return jobBuilderFactory.get("job")
                .flow(stepA()).on("FAILED").to(stepC())
                .from(stepA()).on("*").to(stepB()).next(stepC())
                .end().build();
    }

    @Bean
    public Step stepA() {
        return stepBuilderFactory.get("stepA").tasklet(new RandomFailTasket("stepA")).build();
    }

    @Bean
    public Step stepB() {
        return stepBuilderFactory.get("stepB").tasklet(new PrintTextTasklet("stepB")).build();
    }

    @Bean
    public Step stepC() {
        return stepBuilderFactory.get("stepC").tasklet(new PrintTextTasklet("stepC")).build();
    }

}

我正在使用以下代码开始工作:

    try {
                    Map<String,JobParameter> parameters = new HashMap<>();
                    JobParameter ccReportIdParameter = new JobParameter("03061980");
                    parameters.put("ccReportId", ccReportIdParameter);

                    jobLauncher.run(job, new JobParameters(parameters));
                } catch (JobExecutionAlreadyRunningException | JobRestartException | …

我是春季批次新手.我已经使用inmemoryrepository配置了我的工作.但似乎仍然使用db来持久工作Metadeta.我的春季批量配置是:

@Configuration公共类BatchConfiguration {

@Autowired
private StepBuilderFactory stepBuilderFactory;

@Autowired
private JobBuilderFactory jobBuilder;

@Bean
public JobLauncher jobLauncher() throws Exception {
    SimpleJobLauncher job =new SimpleJobLauncher();
    job.setJobRepository(getJobRepo());
    job.afterPropertiesSet();
    return job;
}


@Bean
public PlatformTransactionManager getTransactionManager() {
    return new ResourcelessTransactionManager();
}

@Bean
public JobRepository getJobRepo() throws Exception {
    return new MapJobRepositoryFactoryBean(getTransactionManager()).getObject();
}




@Bean
public Step step1(JdbcBatchItemWriter<Person> writer) throws Exception {
    return stepBuilderFactory.get("step1")
        .<Person, Person> chunk(10)
        .reader(reader())
        .processor(processor())
        .writer(writer).repository(getJobRepo())
        .build();
}

 @Bean
public Job job( @Qualifier("step1") Step step1) throws Exception {
    return jobBuilder.get("myJob").start(step1).repository(getJobRepo()).build();
}

}

如何解决上述问题？

我有一个单步 springbatch 应用程序。工作内容如下：

@Bean
public Job databaseCursorJob(@Qualifier("databaseCursorStep") Step exampleJobStep,
                             JobBuilderFactory jobBuilderFactory) {
    return jobBuilderFactory.get("databaseCursorJob")
            .incrementer(new RunIdIncrementer())
            .flow(exampleJobStep)
            .end()
            .build();
}

我从 springboot 应用程序开始这项工作。今天下午，我尝试在工作中添加第二步。大致如下：

@Bean
public Job databaseCursorJob(@Qualifier("databaseCursorStep") Step exampleJobStep,
                             JobBuilderFactory jobBuilderFactory) {
    return jobBuilderFactory.get("databaseCursorJob")
            .incrementer(new RunIdIncrementer())
            .flow(exampleJobStep).next(partitionStep())
            .end()
            .build();
}

换句话说，只需添加“next(partitionStep())”。但是，自从我这样做以来，作业就完成了，而没有执行任何步骤（请参见下面的 shell 输出）。事实上，即使在删除第二步并返回到原来的作业，它拒绝执行该步骤。在尝试添加第二步之前，我从未遇到过这个问题。我什至重新启动了我的虚拟机，它仍然跳过该步骤。我已经死在水里了，直到我解决了这个问题。感谢任何见解。谢谢。

2020-09-01 14:49:00.260  INFO 6913 --- [           main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat started on port(s): 8087 (http) with context path ''
2020-09-01 14:49:00.263  INFO 6913 --- [           main] f.p.r.Application    : Started Application in 7.752 seconds (JVM running for 9.092) …

我正在使用Spring-Batch v3.0.0进行批量导入.有一个StepScope和一个JobScope.我怎么知道哪一个合适？

例如,如果我定义一个自定义ItemReader或ItemWriter应该使用特定的自定义EntityManager,它可能如下所示:

@Bean
@Scope("step") //@Scope("job") //custom scope required to inject #jobParameters
public JpaItemWriter<T> jpaItemWriter(EntityManagerFactory emf) {
    JpaItemWriter<T> writer = new JpaItemWriter<T>();
    writer.setEntityManagerFactory(emf);
    return writer;
}

但是哪个范围就在这里？为什么？

带step范围的执行有效,但我觉得itemWrite应该是job范围的,这样它们就不会在每一步都重新创建.

我尝试切换step到job,但是会抛出以下错误: Exception in thread "main" java.lang.IllegalStateException: No Scope registered for scope 'job'

我想配置spring-batch为csv按顺序读取特定文件夹中的所有文件.

以下方法不起作用,因为委托将尝试打开一个名为的文件*.csv,这当然是无效的.我有什么要改变的？

    @Bean
    public ItemReader<String> reader() {
        MultiResourceItemReader<String> reader = new MultiResourceItemReader<>();
        reader.setResources(new Resource[] {new FileSystemResource("/myfolder/*.csv")});
        reader.setDelegate(new FlatFileItemReader<>(..));
        return reader;
}

等效的xml配置如下所示,如何将其重写为仅限java的配置？

<bean id="reader" class="org.springframework.batch.item.file.MultiResourceItemReader">
        <property name="resources" value="/mypfolder/*.csv"/>
        <property name="delegate" ref="flatFileItemReader"/>
</bean>