Kinesis 分析 Flink

本单元中更高层次的构件的API是实验性的并正在积极开发中. 在今后任何版本中,它们都须作非背向相容的修改或去除。 这些是 不适用 语义版本 模型和断开的更改 在发行笔记中宣布. 这意味着虽然你可能使用它们,但可能需要更新 当更新到此软件包的更新版本时,您的源代码.

此软件包提供了创建 Kinesis 分析 Flink 的构造 应用。 要更多地了解使用管理 Flink 应用程序的情况,请参见 页:1 AWS 开发者 指南。 。 。 .

创建 Flink 应用程序

要创建新的 Flink 应用程序,请使用 Application 构造 :

简单的 flink 应用程序

这个 code 属性可以使用 fromAsset 如上文所示,一个当地罐头 s3 中的文件或 fromBucket 以引用 s3 中的文件 .

用桶里的代码链接应用程序

这个 propertyGroups 属性提供了任意运行时间的路由 属性到您的 Flink 应用程序。 你可以用这个 aws-kinesisanatics-runtime 库到 拿来 属性。 。 。 .

declare const bucket: s3.Bucket;
const flinkApp = new flink.Application(this, 'Application', {
  propertyGroups: {
    FlinkApplicationProperties: {
      inputStreamName: 'my-input-kinesis-stream',
      outputStreamName: 'my-output-kinesis-stream',
    },
  },
  // ...
  runtime: flink.Runtime.FLINK_2_2,
  code: flink.ApplicationCode.fromBucket(bucket, 'my-app.jar'),
});

Flink 应用程序也有通过参数的特定配置 当 Flink 任务开始。 其中包括检查站的参数, 示意图、监测和平行性.

declare const bucket: s3.Bucket;
const flinkApp = new flink.Application(this, 'Application', {
  code: flink.ApplicationCode.fromBucket(bucket, 'my-app.jar'),
  runtime: flink.Runtime.FLINK_2_2,
  checkpointingEnabled: true, // default is true
  checkpointInterval: Duration.seconds(30), // default is 1 minute
  minPauseBetweenCheckpoints: Duration.seconds(10), // default is 5 seconds
  logLevel: flink.LogLevel.ERROR, // default is INFO
  metricsLevel: flink.MetricsLevel.PARALLELISM, // default is APPLICATION
  autoScalingEnabled: false, // default is true
  parallelism: 32, // default is 1
  parallelismPerKpu: 2, // default is 1
  snapshotsEnabled: false, // default is true
  logGroup: new logs.LogGroup(this, 'LogGroup'), // by default, a new LogGroup will be created
});

Flink应用程序可以选择在VPC中部署:

declare const bucket: s3.Bucket;
declare const vpc: ec2.Vpc;
const flinkApp = new flink.Application(this, 'Application', {
  code: flink.ApplicationCode.fromBucket(bucket, 'my-app.jar'),
  runtime: flink.Runtime.FLINK_2_2,
  vpc,
});