Java 类名:com.alibaba.alink.operator.batch.sink.RedisRowSinkBatchOp
Python 类名:RedisRowSinkBatchOp
将一个批式数据,按行写到Redis里,键和值可以是多列。
在使用时,需要先下载插件,详情请看https://www.yuque.com/pinshu/alink_guide/czg4cx
| 名称 | 中文名称 | 描述 | 类型 | 是否必须? | 取值范围 | 默认值 |
|---|---|---|---|---|---|---|
| pluginVersion | 插件版本号 | 插件版本号 | String | ✓ | ||
| clusterMode | 集群模式 | 是集群模式还是单机模式 | Boolean | false | ||
| databaseIndex | 数据库索引号 | 数据库索引号 | Long | |||
| keyCols | 多键值列 | 多键值列 | String[] | null | ||
| pipelineSize | 流水线大小 | Redis 发送命令流水线的大小 | Integer | 1 | ||
| redisIPs | Redis IP | Redis 集群的 IP/端口 | String[] | |||
| redisPassword | Redis 密码 | Redis 服务器密码 | String | |||
| timeout | 超时 | 关闭连接的超时时间 | Integer | |||
| valueCols | 多数值列 | 多数值列 | String[] | null |
** 以下代码仅用于示意,可能需要修改部分代码或者配置环境后才能正常运行!**
redisIP = "127.0.0.1:6379"
df = pd.DataFrame([
["football", 1.0],
["football", 2.0],
["football", 3.0],
["basketball", 4.0],
["basketball", 5.0],
["tennis", 6.0],
["tennis", 7.0],
["pingpang", 8.0],
["pingpang", 9.0],
["baseball", 10.0]])
batchData = BatchOperator.fromDataframe(df, schemaStr='id string,val double')
batchData.link(RedisRowSinkBatchOp()\
.setRedisIPs(redisIP)\
.setKeyCols(["id"])\
.setValueCols(["val"])\
.setPluginVersion("2.9.0"))
BatchOperator.execute()
package com.alibaba.alink.operator.batch.sink;
import org.apache.flink.types.Row;
import com.alibaba.alink.operator.batch.BatchOperator;
import com.alibaba.alink.operator.batch.source.MemSourceBatchOp;
import com.alibaba.alink.testutil.AlinkTestBase;
import org.junit.Test;
import java.util.Arrays;
import java.util.List;
public class RedisRowSinkBatchOpTest extends AlinkTestBase {
@Test
public void test() throws Exception {
String redisIP = "127.0.0.1:6379";
List <Row> df = Arrays.asList(
Row.of("football", 1.0),
Row.of("football", 2.0),
Row.of("football", 3.0),
Row.of("basketball", 4.0),
Row.of("basketball", 5.0),
Row.of("tennis", 6.0),
Row.of("tennis", 7.0),
Row.of("pingpang", 8.0),
Row.of("pingpang", 9.0),
Row.of("baseball", 10.0)
);
BatchOperator <?> data = new MemSourceBatchOp(df, "id string,val double");
RedisRowSinkBatchOp sink = new RedisRowSinkBatchOp()
.setPluginVersion("2.9.0")
.setRedisIPs(redisIP)
.setKeyCols("id")
.setValueCols("val");
data.link(sink);
BatchOperator.execute();
}
}