导出到Redis (RedisRowSinkStreamOp)

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

Python 类名：RedisRowSinkStreamOp

功能介绍

将一个流式数据，按行写到Redis里，键和值可以是多列。

在使用时，需要先下载插件，详情请看https://www.yuque.com/pinshu/alink_guide/czg4cx

参数说明

名称 中文名称 描述 类型 是否必须？ 取值范围 默认值
pluginVersion 插件版本号 插件版本号 String ✓
clusterMode Not available! Not available! Boolean false
databaseIndex Not available! Not available! Long
keyCols 多键值列 多键值列 String[] null
pipelineSize Not available! Not available! Integer 1
redisIPs Not available! Not available! String[]
redisPassword Not available! Not available! String
timeout Not available! Not available! Integer
valueCols 多数值列 多数值列 String[] null

代码示例

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

Python 代码

  1. redisIP = "127.0.0.1:6379"
  3. df = pd.DataFrame([
  4.     ["football", 1.0],
  5.     ["football", 2.0],
  6.     ["football", 3.0],
  7.     ["basketball", 4.0],
  8.     ["basketball", 5.0],
  9.     ["tennis", 6.0],
  10.     ["tennis", 7.0],
  11.     ["pingpang", 8.0],
  12.     ["pingpang", 9.0],
  13.     ["baseball", 10.0]])
  15. streamData = StreamOperator.fromDataframe(df, schemaStr='id string,val double')
  17. streamData.link(RedisRowSinkStreamOp()\
  18.             .setRedisIPs(redisIP)\
  19.             .setKeyCols(["id"])\
  20.             .setValueCols(["val"])\
  21.             .setPluginVersion("2.9.0"))
  23. StreamOperator.execute()

Java 代码

  1. package com.alibaba.alink.operator.stream.sink;
  3. import org.apache.flink.types.Row;
  5. import com.alibaba.alink.operator.batch.BatchOperator;
  6. import com.alibaba.alink.operator.batch.source.MemSourceBatchOp;
  7. import com.alibaba.alink.testutil.AlinkTestBase;
  8. import org.junit.Test;
  10. import java.util.Arrays;
  11. import java.util.List;
  13. public class RedisRowSinkStreamOpTest extends AlinkTestBase {
  14.     @Test
  15.     public void test() throws Exception {
  16.         String redisIP = "*";
  17.         int redisPort = 0;
  19.         List <Row> df = Arrays.asList(
  20.             Row.of("football", 1.0),
  21.             Row.of("football", 2.0),
  22.             Row.of("football", 3.0),
  23.             Row.of("basketball", 4.0),
  24.             Row.of("basketball", 5.0),
  25.             Row.of("tennis", 6.0),
  26.             Row.of("tennis", 7.0),
  27.             Row.of("pingpang", 8.0),
  28.             Row.of("pingpang", 9.0),
  29.             Row.of("baseball", 10.0)
  30.         );
  32.         StreamOperator <?> data = new MemSourceStreamOp(df, "id string,val double");
  34.         RedisRowSinkStreamOp sink = new RedisRowSinkStreamOp()
  35.             .setRedisIPs(redisIP)
  36.             .setKeyCols("id")
  37.             .setValueCols("val")
  38.             .setPluginVersion("2.9.0");
  40.         data.link(sink);
  42.         StreamOperator.execute();
  43.     }
  45. }