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

功能介绍

支持数据查找功能，支持多个key的查找，并将查找后的结果中的value列添加到待查询数据后面。与SQl语法中的inner join功能类似，当不存在重复的key时
LookupBatchOp().setMapKeyCols(“key_col_A”).setMapValueCols(“value_col”).setSelectedCols(“key_col_B”).linkFrom(A, B)与
“SELECT A.value_col FROM A INNER JOIN B ON A.key_col_A = B.key_col_B”，但是需要注意：当数据B中存在多行相同的key时，只保留一个value，不会找到所有的value。
Table A

| key_col_A | value_col | | —- | —- |

| Bob | 98 |

| Tom | 72 |

Table B

| key_col_B | age | | —- | —- |

| Bob | 11 |

| Denny | 10 |

查找结果

| key_col_B | age | value_col | | —- | —- | —- |

| Bob | 11 | 98 |

| Denny | 10 | null |

参数说明

名称 中文名称 描述 类型 是否必须？ 取值范围 默认值
selectedCols 选择的列名 计算列对应的列名列表 String[] ✓
mapKeyCols Key列名 模型中对应的查找等值的列名 String[] null
mapValueCols Values列名 模型中需要拼接到样本中的列名 String[] null
modelFilePath 模型的文件路径 模型的文件路径 String null
outputCols 输出结果列列名数组 输出结果列列名数组，可选，默认null String[] null
reservedCols 算法保留列名 算法保留列 String[] null
numThreads 组件多线程线程个数 组件多线程线程个数 Integer 1
modelStreamUpdateMethod 模型更新方法 模型更新方法，可选COMPLETE（全量更新）或者 INCREMENT（增量更新） String “COMPLETE”, “INCREMENT” “COMPLETE”

代码示例

Python 代码

  1. from pyalink.alink import *
  3. import pandas as pd
  5. useLocalEnv(1)
  7. df = pd.DataFrame([
  8.     ["10", 2.0], ["1", 2.0], ["-3", 2.0], ["5", 1.0]
  9. ])
  11. inOp = BatchOperator.fromDataframe(df, schemaStr='f0 string, f1 double')
  13. df2 = pd.DataFrame([
  14.     ["1", "value1"], ["2", "value2"], ["5", "value5"]
  15. ])
  16. modelOp = BatchOperator.fromDataframe(df2, schemaStr="key_col string, value_col string")
  18. LookupBatchOp().setMapKeyCols(["key_col"]).setMapValueCols(["value_col"]) \
  19.     .setSelectedCols(["f0"]).linkFrom(modelOp, inOp).print()

Java 代码

  1. import org.apache.flink.types.Row;
  3. import com.alibaba.alink.operator.batch.BatchOperator;
  4. import com.alibaba.alink.operator.batch.dataproc.LookupBatchOp;
  5. import com.alibaba.alink.operator.batch.source.MemSourceBatchOp;
  6. import org.junit.Test;
  8. import java.util.Arrays;
  9. import java.util.List;
  11. public class LookupBatchOpTest {
  12.     @Test
  13.     public void testLookupBatchOp() throws Exception {
  14.         List <Row> df = Arrays.asList(
  15.             Row.of("10", 2.0), Row.of("1", 2.0), Row.of("-3", 2.0), Row.of("5", 1.0)
  16.         );
  17.         BatchOperator <?> inOp = new MemSourceBatchOp(df, "f0 string, f1 double");
  18.         List <Row> df2 = Arrays.asList(
  19.             Row.of("1", "value1"), Row.of("2", "value2"), Row.of("5", "value5")
  20.         );
  21.         BatchOperator <?> modelOp = new MemSourceBatchOp(df2, "key_col string, value_col string");
  22.         new LookupBatchOp().setMapKeyCols("key_col").setMapValueCols("value_col")
  23.             .setSelectedCols("f0").linkFrom(modelOp, inOp).print();
  24.     }
  25. }

运行结果

| f0 | f1 | value_col | | —- | —- | —- |

| 10 | 2.0 | null |

| 1 | 2.0 | value1 |

| -3 | 2.0 | null |

| 5 | 1.0 | value5 |