使用Ignite Compute从Snowflake加载数据遇Lambda序列化异常求助

问题描述 #

尝试通过Ignite Compute将Snowflake数据加载到Ignite集群，所有服务节点和客户端均已部署Snowflake驱动，但执行Compute作业时抛出序列化异常，提示Snowflake客户端内部的Lambda无法序列化。已尝试URIDeploymentSPI但问题未解决，需求如下：

1. 是否可以禁用Lambda表达式序列化？
2. 有没有性能优于Data Streamer的数据加载方式？

异常信息 #

Caused by: class org.apache.ignite.binary.BinaryObjectException: Lambda is not serializable: class net.snowflake.client.core.QueryContextCache$$Lambda$975/0x00000008011d6018 
at org.apache.ignite.internal.binary.BinaryClassDescriptor.<init>(BinaryClassDescriptor.java:325) 
at org.apache.ignite.internal.binary.BinaryClassDescriptor.<init>(BinaryClassDescriptor.java:156) 
at org.apache.ignite.internal.binary.BinaryContext.createDescriptorForClass(BinaryContext.java:675) 
at org.apache.ignite.internal.binary.BinaryContext.descriptorForClass(BinaryContext.java:633) 
at org.apache.ignite.internal.binary.BinaryWriterExImpl.marshal0(BinaryWriterExImpl.java:182) 
at org.apache.ignite.internal.binary.BinaryWriterExImpl.marshal(BinaryWriterExImpl.java:165) 
at org.apache.ignite.internal.binary.BinaryWriterExImpl.marshal(BinaryWriterExImpl.java:152) 
at org.apache.ignite.internal.binary.BinaryWriterExImpl.doWriteObject(BinaryWriterExImpl.java:519) 
at org.apache.ignite.internal.binary.BinaryWriterExImpl.doWriteObjectArray(BinaryWriterExImpl.java:769) 
at org.apache.ignite.internal.binary.BinaryWriterExImpl.writeObjectArrayField(BinaryWriterExImpl.java:1323) 
at org.apache.ignite.internal.binary.BinaryFieldAccessor$DefaultFinalClassAccessor.write0(BinaryFieldAccessor.java:650) 
at org.apache.ignite.internal.binary.BinaryFieldAccessor.write(BinaryFieldAccessor.java:157) 
... 93 more

一、Lambda序列化问题解决方案 #

不能直接禁用Lambda序列化，问题根源是Snowflake客户端内部的匿名Lambda被包含在Compute作业的闭包中，需要序列化传输到服务节点，但动态生成的Lambda类不支持Ignite的二进制序列化。可通过以下方式解决：

1. 服务节点本地初始化Snowflake连接 #

不在客户端创建Snowflake连接后传入Compute闭包，而是在Compute任务的call()方法内部，在服务节点本地初始化连接、执行查询，避免传输包含Lambda的客户端对象：

IgniteCompute compute = ignite.compute();

// 自定义可序列化Callable，在服务节点本地执行查询
compute.call(new SerializableCallable<List<MyData>>() {
    @Override
    public List<MyData> call() throws Exception {
        try (Connection conn = DriverManager.getConnection("jdbc:snowflake://...", "user", "pass")) {
            String sql = "SELECT * FROM snowflake_table LIMIT 1000";
            try (Statement stmt = conn.createStatement(); ResultSet rs = stmt.executeQuery(sql)) {
                List<MyData> data = new ArrayList<>();
                while (rs.next()) {
                    data.add(new MyData(rs.getInt("id"), rs.getString("name")));
                }
                return data;
            }
        }
    }
});

// 定义可序列化的Callable接口
interface SerializableCallable<V> extends Callable<V>, Serializable {}

2. 替换Lambda为可序列化的自定义类 #

避免使用Lambda表达式，改用实现Callable和Serializable的具体类，确保所有成员变量可序列化：

public class SnowflakeLoadTask implements Callable<List<MyData>>, Serializable {
    private String jdbcUrl;
    private String user;
    private String pass;
    private String sql;

    public SnowflakeLoadTask(String jdbcUrl, String user, String pass, String sql) {
        this.jdbcUrl = jdbcUrl;
        this.user = user;
        this.pass = pass;
        this.sql = sql;
    }

    @Override
    public List<MyData> call() throws Exception {
        try (Connection conn = DriverManager.getConnection(jdbcUrl, user, pass)) {
            try (Statement stmt = conn.createStatement(); ResultSet rs = stmt.executeQuery(sql)) {
                List<MyData> data = new ArrayList<>();
                while (rs.next()) {
                    data.add(new MyData(rs.getInt("id"), rs.getString("name")));
                }
                return data;
            }
        }
    }
}

// 使用方式
IgniteCompute compute = ignite.compute();
List<MyData> result = compute.call(new SnowflakeLoadTask("jdbc:snowflake://...", "user", "pass", "SELECT * FROM table"));

3. 升级Snowflake客户端版本 #

部分旧版本Snowflake JDBC驱动中，QueryContextCache内部的Lambda存在序列化缺陷，升级到最新稳定版（如3.13+）可能修复该问题。

二、高性能数据加载替代方案 #

如果Data Streamer性能未达预期，可尝试以下方案：

1. 分布式CacheStore批量加载 #

实现自定义CacheStore，利用Ignite分布式加载能力，每个节点负责加载分片数据，并行处理：

public class SnowflakeCacheStore extends CacheStoreAdapter<Integer, MyData> {
    @Override
    public void loadCache(IgniteBiInClosure<Integer, MyData> clo, Object... args) throws CacheLoaderException {
        try (Connection conn = DriverManager.getConnection("jdbc:snowflake://...", "user", "pass")) {
            // 根据节点分片范围过滤数据，实现分布式加载
            String sql = "SELECT id, name FROM snowflake_table WHERE id BETWEEN ? AND ?";
            try (PreparedStatement stmt = conn.prepareStatement(sql)) {
                int start = getNodeStartId(); // 自定义方法获取当前节点负责的ID起始值
                int end = getNodeEndId(); // 自定义方法获取当前节点负责的ID结束值
                stmt.setInt(1, start);
                stmt.setInt(2, end);
                try (ResultSet rs = stmt.executeQuery()) {
                    while (rs.next()) {
                        MyData data = new MyData(rs.getInt("id"), rs.getString("name"));
                        clo.apply(data.getId(), data);
                    }
                }
            }
        } catch (SQLException e) {
            throw new CacheLoaderException(e);
        }
    }

    // 按需实现其他CacheStore方法...
}

配置缓存启用CacheStore：

<bean class="org.apache.ignite.configuration.CacheConfiguration">
    <property name="name" value="myCache"/>
    <property name="cacheStoreFactory">
        <bean class="org.apache.ignite.cache.store.factory.CacheStoreFactory">
            <property name="factory" value="com.example.SnowflakeCacheStore"/>
        </bean>
    </property>
    <property name="readThrough" value="true"/>
    <property name="writeThrough" value="false"/>
</bean>

触发加载：ignite.cache("myCache").loadCache(null)

2. 优化Data Streamer配置 #

若坚持使用Data Streamer，调整以下配置提升性能：

IgniteDataStreamer<Integer, MyData> streamer = ignite.dataStreamer("myCache");
// 设置每个节点的缓冲区大小
streamer.perNodeBufferSize(1024);
// 设置每个节点的并行操作数
streamer.perNodeParallelOperations(8);
// 自动刷新频率（毫秒），避免缓冲区满才提交
streamer.autoFlushFrequency(1000);
// 跳过持久化层（无需写回数据库时启用）
streamer.skipStore(true);
// 启用异步写后台线程
streamer.writeBehindEnabled(true);
streamer.writeBehindFlushSize(4096);
streamer.writeBehindFlushFrequency(2000);

// 批量添加数据
List<MyData> batch = fetchSnowflakeBatch();
for (MyData data : batch) {
    streamer.addData(data.getId(), data);
}
streamer.close();

3. 使用Ignite SQL IMPORT命令（Ignite 2.10+） #

利用Ignite分布式SQL导入能力，直接从Snowflake JDBC数据源并行加载数据：

IMPORT INTO my_cache (id, name)
FROM JDBC 'jdbc:snowflake://account.snowflakecomputing.com/?user=xxx&password=xxx&warehouse=wh&db=db&schema=schema'
SELECT id, name FROM snowflake_table

该命令会在Ignite集群分布式执行，每个节点拉取部分数据，性能远高于单节点Data Streamer。

4. 列式存储文件批量加载 #

将Snowflake数据导出为Parquet/ORC等列式存储格式，上传到分布式文件系统（如HDFS），再用Ignite文件系统集成并行加载：

IgniteFileSystem fs = ignite.fileSystem("hfs");
Collection<IgniteFsPath> files = fs.listPaths(new IgniteFsPath("/snowflake_data"));
for (IgniteFsPath path : files) {
    ignite.compute().run(() -> {
        try (InputStream in = fs.open(path)) {
            ParquetReader<MyData> reader = ParquetReader.builder(new MyDataReadSupport(), Paths.get(path.toString())).build();
            MyData data;
            while ((data = reader.read()) != null) {
                ignite.cache("myCache").put(data.getId(), data);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    });
}

内容的提问来源于stack exchange，提问作者Rakesh Balguri