使用 Redshift 作为 Spring 批处理作业存储库和 Redshift 中 SEQUENCE 的替代品答案

【问题标题】：Using Redshfit as Spring batch Job Repository and alternatives to SEQUENCE in Redshfit使用 Redshift 作为 Spring 批处理作业存储库和 Redshift 中 SEQUENCE 的替代品
【发布时间】：2013-12-26 18:56:07
【问题描述】：

我的项目中的一个要求是将 Spring Batch 模式放置在 amazon redshift db 上。
我打算从 schema-postgresql.sql 作为基线开始，因为 redshift 是基于 postgres 的。

查看 spring 批处理源代码，您似乎需要做一些事情来完成这项工作：

扩展 JobRepositoryFactoryBean、DefaultDataFieldMaxValueIncrementerFactory。
添加我自己的扩展 AbstractSequenceMaxValueIncrementer 的 RedshfitMaxValueIncrementer

查看redshift datatypes，除了用于创建作业、执行、步骤执行 ID 的序列之外，转换模式脚本似乎没有任何问题。

您对缺失序列的最佳解决方法有什么建议？

将这些列指定为IDENTITY 列。从红移的角度来看，这是最简单的方法。这可能是有问题的 DataFieldMaxValueIncrementer.nextLongValue() 返回 long 而不是很长，我们需要返回 null 并让 IDENTITY 为我们完成这项工作
基于类似 select max(STEP_EXECUTION_ID) from BATCH_STEP_EXECUTION 的实现并做一些类似于 MySQLMaxValueIncrementer 的扩展 AbstractColumnMaxValueIncrementer
仅在 java 代码中创建序列；使用类似的工具休眠使用
上面没有提到的一种方法

【问题讨论】：

有人吗？有什么想法吗？
你解决过这个问题吗？
在我有机会完成此操作之前，我们切换到了不同的数据库。

标签： spring amazon-web-services spring-batch amazon-redshift

【解决方案1】：

这就是我如何至少让那部分（显然）工作：

在DefaultBatchConfigurer 的子类中，我添加了以下代码：

@Override
protected JobRepository createJobRepository() throws Exception
{
    JobRepositoryFactoryBean factory = new JobRepositoryFactoryBean();
    factory.setDataSource(dataSource);
    factory.setTransactionManager(getTransactionManager());
    factory.setIncrementerFactory(new RedshiftIncrementerFactory(dataSource));
    factory.afterPropertiesSet();
    return factory.getObject();
}

工厂对象的样子

public class RedshiftIncrementerFactory implements DataFieldMaxValueIncrementerFactory
{
    private DataSource dataSource;

    public RedshiftIncrementerFactory(DataSource ds)
    {
        this.dataSource = ds;
    }

    @Override
    public DataFieldMaxValueIncrementer getIncrementer(String databaseType, String incrementerName)
    {
        return new RedshiftIncrementer(dataSource, incrementerName);
    }

    @Override
    public boolean isSupportedIncrementerType(String databaseType)
    {
        return POSTGRES.toString().equals(databaseType);
    }

    @Override
    public String[] getSupportedIncrementerTypes()
    {
        return new String[]{POSTGRES.toString()};
    }

}

然后，最后是增量器本身：

public class RedshiftIncrementer extends AbstractSequenceMaxValueIncrementer
{
    public RedshiftIncrementer(DataSource dataSource, String incrementorName)
    {
        super(dataSource, incrementorName);
    }

    // I need to run two queries here, since Redshift doesn't support sequences
    @Override
    protected long getNextKey() throws DataAccessException {
        Connection con = DataSourceUtils.getConnection(getDataSource());
        Statement stmt = null;
        ResultSet rs = null;
        try {
            stmt = con.createStatement();
            DataSourceUtils.applyTransactionTimeout(stmt, getDataSource());
            String table = getIncrementerName();
            stmt.executeUpdate("UPDATE " + table + " SET ID = ID + 1");
            rs = stmt.executeQuery("SELECT ID FROM " + table + " WHERE UNIQUE_KEY='0'");
            if (rs.next()) {
                return rs.getLong(1);
            }
            else {
                throw new DataAccessResourceFailureException("Sequence query did not return a result");
            }
        }
        catch (SQLException ex) {
            throw new DataAccessResourceFailureException("Could not obtain sequence value", ex);
        }
        finally {
            JdbcUtils.closeResultSet(rs);
            JdbcUtils.closeStatement(stmt);
            DataSourceUtils.releaseConnection(con, getDataSource());
        }
    }

    @Override
    protected String getSequenceQuery()
    {
        // No longer used
        return null;
    }
}

这至少允许作业开始。但是，Redshift 还存在其他问题，我将在其他地方详细说明。

【讨论】：