Amazon 管理 Apache Kafka 建构库
本单元中更高层次的构件的API是实验性的并正在积极开发中. 在今后任何版本中,它们都须作非背向相容的修改或去除。 这些是 不适用 语义版本 模型和断开的更改 在发行笔记中宣布. 这意味着虽然你可能使用它们,但可能需要更新 当更新到此软件包的更新版本时,您的源代码.
亚马逊 MSK 是一种完全管理的服务,它使您可以轻松地构建和运行使用Apache Kafka处理流数据的程序.
以下例子创建了MSK Cluster.
declare const vpc: ec2.Vpc;
const cluster = new msk.Cluster(this, 'Cluster', {
clusterName: 'myCluster',
kafkaVersion: msk.KafkaVersion.V4_1_X_KRAFT,
vpc,
});
允许连接
为了控制谁可以访问集群,请使用 .connections 属性。 关于MSK使用的端口列表,请参见: MSK 文档。 。 。 .
declare const vpc: ec2.Vpc;
const cluster = new msk.Cluster(this, 'Cluster', {
clusterName: 'myCluster',
kafkaVersion: msk.KafkaVersion.V4_1_X_KRAFT,
vpc,
});
cluster.connections.allowFrom(
ec2.Peer.ipv4('1.2.3.4/8'),
ec2.Port.tcp(2181),
);
cluster.connections.allowFrom(
ec2.Peer.ipv4('1.2.3.4/8'),
ec2.Port.tcp(9094),
);
集群终点
您可以使用以下属性来获取 Kafka 经纪人或 Zookeeper 节点的列表
declare const cluster: msk.Cluster;
new CfnOutput(this, 'BootstrapBrokers', { value: cluster.bootstrapBrokers });
new CfnOutput(this, 'BootstrapBrokersTls', { value: cluster.bootstrapBrokersTls });
new CfnOutput(this, 'BootstrapBrokersSaslScram', { value: cluster.bootstrapBrokersSaslScram });
new CfnOutput(this, 'BootstrapBrokerStringSaslIam', { value: cluster.bootstrapBrokersSaslIam });
new CfnOutput(this, 'ZookeeperConnection', { value: cluster.zookeeperConnectionString });
new CfnOutput(this, 'ZookeeperConnectionTls', { value: cluster.zookeeperConnectionStringTls });
导入已存在的集群
要将已有的 MSK 集群导入到您的 CDK 应用程序中, 请使用 .fromClusterArn() 方法.
const cluster = msk.Cluster.fromClusterArn(this, 'Cluster',
'arn:aws:kafka:us-west-2:1234567890:cluster/a-cluster/11111111-1111-1111-1111-111111111111-1',
);
客户端认证
MSK 支持 以下认证机制.
电话
用 TLS 设置客户端认证 certificateAuthorityArns 以引用您的 ACM 私人 CA. 更多关于私人CA的信息.
import * as acmpca from 'aws-cdk-lib/aws-acmpca';
declare const vpc: ec2.Vpc;
const cluster = new msk.Cluster(this, 'Cluster', {
clusterName: 'myCluster',
kafkaVersion: msk.KafkaVersion.V4_1_X_KRAFT,
vpc,
encryptionInTransit: {
clientBroker: msk.ClientBrokerEncryption.TLS,
},
clientAuthentication: msk.ClientAuthentication.tls({
certificateAuthorities: [
acmpca.CertificateAuthority.fromCertificateAuthorityArn(
this,
'CertificateAuthority',
'arn:aws:acm-pca:us-west-2:1234567890:certificate-authority/11111111-1111-1111-1111-111111111111',
),
],
}),
});
SASL/SRAM 软件
启用客户端认证 SASL/SRAM 软件数字 :
declare const vpc: ec2.Vpc;
const cluster = new msk.Cluster(this, 'cluster', {
clusterName: 'myCluster',
kafkaVersion: msk.KafkaVersion.V4_1_X_KRAFT,
vpc,
encryptionInTransit: {
clientBroker: msk.ClientBrokerEncryption.TLS,
},
clientAuthentication: msk.ClientAuthentication.sasl({
scram: true,
}),
});
行政会议
启用客户端认证 行政会议数字 :
declare const vpc: ec2.Vpc;
const cluster = new msk.Cluster(this, 'cluster', {
clusterName: 'myCluster',
kafkaVersion: msk.KafkaVersion.V4_1_X_KRAFT,
vpc,
encryptionInTransit: {
clientBroker: msk.ClientBrokerEncryption.TLS,
},
clientAuthentication: msk.ClientAuthentication.sasl({
iam: true,
}),
});
SASL/IAM + TLS
启用客户端认证 行政会议
通过设置 TLS 并启用客户端认证 certificateAuthorityArns 以引用您的 ACM 私人 CA. 更多关于私人CA的信息.
import * as acmpca from 'aws-cdk-lib/aws-acmpca';
declare const vpc: ec2.Vpc;
const cluster = new msk.Cluster(this, 'Cluster', {
clusterName: 'myCluster',
kafkaVersion: msk.KafkaVersion.V4_1_X_KRAFT,
vpc,
encryptionInTransit: {
clientBroker: msk.ClientBrokerEncryption.TLS,
},
clientAuthentication: msk.ClientAuthentication.saslTls({
iam: true,
certificateAuthorities: [
acmpca.CertificateAuthority.fromCertificateAuthorityArn(
this,
'CertificateAuthority',
'arn:aws:acm-pca:us-west-2:1234567890:certificate-authority/11111111-1111-1111-1111-111111111111',
),
],
}),
});
日志
您可以将 Apache Kafka 代理日志发送给以下一个或多个目标类型: 亚马逊云表日志 亚马逊 S3 亚马逊数据 Firehose.
要配置要发送给 S3 桶的日志, 请在 logging 相会.
declare const vpc: ec2.Vpc;
declare const bucket: s3.IBucket;
const cluster = new msk.Cluster(this, 'cluster', {
clusterName: 'myCluster',
kafkaVersion: msk.KafkaVersion.V4_1_X_KRAFT,
vpc,
logging: {
s3: {
bucket,
},
},
});
S3 目的地配置后, AWS 会自动创建 S3 桶策略
允许服务将日志写入桶中。 这使得以后无法更新
那个桶政策。 要让 CDK 创建桶策略, 以便将来可以更新,
页:1 @aws-cdk/aws-s3:createDefaultLoggingPolicy 特性旗 可以使用。 这个可以设定
输入 cdk.json 文档.
{
"context": {
"@aws-cdk/aws-s3:createDefaultLoggingPolicy": true
}
}
存储模式
您可以使用 storageMode 属性.
分层存储是亚马逊MSK的低成本存储级,其分级到几乎无限存储, 使建立流数据应用程序具有成本效益.
访问 分级存储 ,以查看兼容的卡夫卡版本列表,并了解更多细节.
declare const vpc: ec2.Vpc;
declare const bucket: s3.IBucket;
const cluster = new msk.Cluster(this, 'cluster', {
clusterName: 'myCluster',
kafkaVersion: msk.KafkaVersion.V4_1_X_KRAFT,
vpc,
storageMode: msk.StorageMode.TIERED,
});
MSK快递经纪人
您可以通过设置 : brokerType 属性改为 BrokerType.EXPRESS. Express Brokers是开发,测试,工作量的低成本选择,不需要标准MSK集群的高可用性保证.
更多信息见 亚马逊 MSK 特快经纪。 。 。 .
说明: 在使用Express经纪时,适用以下限制:
- Apache Kafka版本必须是 3.6.x, 3.8.x, 3.9.x, 或 4.2.x
- 您必须指定
instanceType - VPC必须至少有3个子网(跨越3个AZ)
ebsStorageInfo不支持storageMode不支持logging不支持- 支持的经纪人大小 :
m7g.xlarge, (中文(简体) ),m7g.2xlarge, (中文(简体) ),m7g.4xlarge, (中文(简体) ),m7g.8xlarge, (中文(简体) ),m7g.12xlarge, (中文(简体) ),m7g.16xlarge
declare const vpc: ec2.Vpc;
const expressCluster = new msk.Cluster(this, 'ExpressCluster', {
clusterName: 'MyExpressCluster',
kafkaVersion: msk.KafkaVersion.V3_8_X,
vpc,
brokerType: msk.BrokerType.EXPRESS,
instanceType: ec2.InstanceType.of(
ec2.InstanceClass.M7G,
ec2.InstanceSize.XLARGE,
),
});
无 MSK 服务器
您也可以通过使用 MSK 无服务器 ServerlessCluster 课.
MSK Serverless是亚马逊MSK的一种集群类型,它使得您可以运行Apache Kafka而无需管理和规模化集群能力.
MSK Serverless要求所有集群的IAM访问控制.
关于更多的信息,见 使用 MSK 无服务器集群。 。 。 .
declare const vpc: ec2.Vpc;
const serverlessCluster = new msk.ServerlessCluster(this, 'ServerlessCluster', {
clusterName: 'MyServerlessCluster',
vpcConfigs: [
{ vpc },
],
});