导出到Redis (RedisStringSinkStreamOp)

Java 类名：com.alibaba.alink.operator.stream.sink.RedisStringSinkStreamOp

Python 类名：RedisStringSinkStreamOp

功能介绍

将一个流式数据，（单列String类型键值）按行写到Redis里。

注意事项

• 与RedisRowSinkStreamOp不同写入的是序列化后的字符串，例如：”A”写入Redis的结果为”A”
• 在使用时，需要先下载插件，详情请看https://www.yuque.com/pinshu/alink_guide/czg4cx

参数说明

代码示例

以下代码仅用于示意，可能需要修改部分代码或者配置环境后才能正常运行！

Python 代码

redisIP = "*"
redisPort = 0
df = pd.DataFrame([
    ["football", "1.0"],
    ["football", "2.0"],
    ["football", "3.0"]])
batchData = StreamOperator.fromDataframe(df, schemaStr='id string,val double')
batchData.link(RedisStringSinkStreamOp()\
            .setRedisIPs(redisIP)\
            .setKeyCol(["id"])\
            .setValueCol(["val"])\
            .setPluginVersion("2.9.0"))
BatchOperator.execute()

Java 代码

package com.alibaba.alink.operator.batch.sink;
import org.apache.flink.types.Row;
import com.alibaba.alink.operator.stream.StreamOperator;
import com.alibaba.alink.operator.stream.source.MemSourceStreamOp;
import com.alibaba.alink.operator.stream.sink.RedisStringSinkStreamOp;
import com.alibaba.alink.testutil.AlinkTestBase;
import org.junit.Test;
import java.util.Arrays;
import java.util.List;
public class RedisStringSinkStreamOpTest extends AlinkTestBase {
    @Test
    public void test() throws Exception {
        String redisIP = "127.0.0.1:6379";
        int redisPort = 0;
        List <Row> df = Arrays.asList(
            Row.of("football", "1.0"),
            Row.of("football", "2.0")
        );
        StreamOperator <?> data = new MemSourceStreamOp(df, "id string,val string");
        RedisStringSinkStreamOp sink = new RedisStringSinkStreamOp()
            .setRedisIPs(redisIP)
            .setKeyCol("id")
            .setValueCol("val")
            .setPluginVersion("2.9.0");
        data.link(sink);
        StreamOperator.execute();
    }
}