Java 类名：com.alibaba.alink.operator.batch.dataproc.LookupRedisBatchOp
Python 类名：LookupRedisBatchOp

功能介绍

支持数据查找功能，支持多个key的查找，并将查找后的结果中的value列添加到待查询数据后面。
功能类似于 LookUpBatchOp ，不同的是被查找的数据存储在 Redis 中。

参数说明

名称 中文名称 描述 类型 是否必须？ 取值范围 默认值
outputSchemaStr Schema Schema。格式为”colname coltype[, colname2, coltype2[, …]]”，例如”f0 string, f1 bigint, f2 double” String ✓
pluginVersion 插件版本号 插件版本号 String ✓
selectedCols 选择的列名 计算列对应的列名列表 String[] ✓
clusterMode Not available! Not available! Boolean false
databaseIndex Not available! Not available! Long
redisIP Not available! Not available! String
redisIPs Not available! Not available! String[]
redisPassword Not available! Not available! String
redisPort Not available! Not available! Integer 6379
reservedCols 算法保留列名 算法保留列 String[] null
timeout Not available! Not available! Integer

代码示例

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

Python 代码

  1. df = pd.DataFrame([
  2.     ["id001", 123, 45.6, "str"]
  3. ])
  5. inOp = BatchOperator.fromDataframe(df, schemaStr='id string, col0 bigint, col1 double, col2 string')
  7. redisIP = "*"
  8. redisPort = 26379
  10. RedisSinkBatchOp()\
  11.     .setRedisIP(redisIP)\
  12.     .setRedisPort(redisPort)\
  13.     .setKeyCols(["id"])\
  14.     .setPluginVersion("2.9.0")\
  15.     .setValueCols(["col0", "col1", "col2"])\
  16.     .linkFrom(inOp)
  18. BatchOperator.execute()
  20. df2 = pd.DataFrame([
  21.     ["id001"]
  22. ])
  23. needToLookup = BatchOperator.fromDataframe(df2, schemaStr="id string")
  25. LookupRedisBatchOp()\
  26.     .setRedisIP(redisIP)\
  27.     .setRedisPort(redisPort)\
  28.     .setPluginVersion("2.9.0")\
  29.     .setSelectedCols(["id"])\
  30.     .setOutputSchemaStr("col0 bigint, col1 double, col2 string")\
  31.     .linkFrom(needToLookup)\
  32.     .lazyPrint(10)
  34. BatchOperator.execute()

Java 代码

  1. import com.alibaba.alink.common.AlinkGlobalConfiguration;
  2. import com.alibaba.alink.operator.batch.BatchOperator;
  3. import com.alibaba.alink.operator.batch.dataproc.LookupRedisBatchOp;
  4. import com.alibaba.alink.operator.batch.sink.RedisSinkBatchOp;
  5. import com.alibaba.alink.operator.batch.source.MemSourceBatchOp;
  6. import com.alibaba.alink.testutil.AlinkTestBase;
  7. import org.junit.Test;
  9. import java.util.Collections;
  11. public class LookupRedisBatchOpTest extends AlinkTestBase {
  12.     @Test
  13.     public void map() throws Exception {
  14.         String redisIP = "*";
  15.         int redisPort = 26379;
  17.         MemSourceBatchOp memSourceBatchOp = new MemSourceBatchOp(
  18.             Collections.singletonList(Row.of("id001", 123L, 45.6, "str")),
  19.             "id string, col0 bigint, col1 double, col2 string"
  20.         );
  22.         new RedisSinkBatchOp()
  23.             .setRedisIP(redisIP)
  24.             .setRedisPort(redisPort)
  25.             .setKeyCols("id")
  26.             .setPluginVersion("2.9.0")
  27.             .setValueCols("col0", "col1", "col2")
  28.             .linkFrom(memSourceBatchOp);
  30.         BatchOperator.execute();
  32.         MemSourceBatchOp needToLookup = new MemSourceBatchOp(
  33.             Collections.singletonList(Row.of("id001")),
  34.             "id string"
  35.         );
  37.         new LookupRedisBatchOp()
  38.             .setRedisIP(redisIP)
  39.             .setRedisPort(redisPort)
  40.             .setPluginVersion("2.9.0")
  41.             .setSelectedCols("id")
  42.             .setOutputSchemaStr("col0 bigint, col1 double, col2 string")
  43.             .linkFrom(needToLookup)
  44.             .lazyPrint(10);
  46.         BatchOperator.execute();
  47.     }
  49. }

运行结果

| id | col0 | col1 | col2 |
|———-+———+————-+———|
| id001 | 123 | 45.6000 | str |