MaxCompute

MaxCompute EndPoint： https://help.aliyun.com/document_detail/34951.html，这里选择的是外网Endpoint： http://service.cn-beijing.maxcompute.aliyun.com/api

Redis

Redis 文档地址为：https://help.aliyun.com/product/26340.html

  1. 创建实例

image.png

  1. 这里创建按量付费本地盘实例
  2. 将 Redis 实例创建进和 Flink 全托管相同的网段内
  3. 获取专有网络地址

image.png

  1. 设置内网网段进白名单，使 Flink 集群可以访问到 Redis 实例

    实例

  2. 代码 ```java package org.example;

import org.apache.flink.ml.api.misc.param.Params; import org.apache.flink.table.catalog.ObjectPath;

import com.alibaba.alink.common.io.catalog.MySqlCatalog; import com.alibaba.alink.common.io.catalog.OdpsCatalog; import com.alibaba.alink.operator.batch.BatchOperator; import com.alibaba.alink.operator.batch.sink.CatalogSinkBatchOp; import com.alibaba.alink.operator.batch.sink.RedisSinkBatchOp; import com.alibaba.alink.operator.batch.source.CatalogSourceBatchOp; import com.alibaba.alink.params.io.HasCatalogObject.CatalogObject; import com.alibaba.alink.params.shared.HasOverwriteSink;

public class HelloAlink {

  1. public static void main(String[] args) throws Exception {
  2.     runSinkRedis();
  3. }
  5. public static void runSinkRedis() throws Exception {
  6.     // 填写好 EndPoint，AccessKeyID，AccessKeySecret, Project。 RunningProject 填写和 Project 相同
  7.     OdpsCatalog odpsCatalog = new OdpsCatalog(
  8.         "odps", "default",
  9.         "0.36.4-public", "AccessKeyID", "AccessKeySecret",
  10.         "Project",
  11.         "EndPoint",
  12.         "Project"
  13.     );
  15.     // schema: sepal_length double, sepal_width double, petal_length double, petal_width double, category string
  16.     CatalogSourceBatchOp source = new CatalogSourceBatchOp()
  17.         .setCatalogObject(
  18.             new CatalogObject(
  19.                 odpsCatalog,
  20.                 // 填写好 Project
  21.                 new ObjectPath("Project", "test_alink_iris"),
  22.                 new Params().set(HasOverwriteSink.OVERWRITE_SINK, true)
  23.             )
  24.         );
  26.     RedisSinkBatchOp sink = new RedisSinkBatchOp()
  27.         .setRedisIPs("ip:port")
  28.         .setRedisPassword("PASSWORD")
  29.         .setClusterMode(true)
  30.         .setDatabaseIndex(0)
  31.         .setKeyCols("sepal_length", "sepal_width", "petal_length", "petal_width")
  32.         .setValueCols("category")
  33.         .setPluginVersion("2.9.0");
  35.     source.link(sink);
  37.     BatchOperator.execute();
  38. }

} ```

  1. 特性
  • 支持集群模式
  • 支持选择数据库索引号
  • 当前存储为 byte[]， 未来版本中会进行扩展