AWS Lambda( TypeScript) - Kafka 工具
用于 AWS Lambda (TypeScript) 的 Powertools 是用于执行无服务器的开发者工具包 最佳做法和增加开发者的速度。 。 。 .
您可以在 TypeScript 和 JavaScript 代码基础中使用软件包.
介绍
卡夫卡消费者公用系统透明地处理信息去区域化,提供直观的开发者经验,并与其他动力工具无缝地融合,用于AWS Lambda生态系统.
使用量
要启动, 根据您想要使用的策略类型, 安装库和相应的库 :
JSON计划:
npm install @aws-lambda-powertools/kafka
对于Avro计划:
npm install @aws-lambda-powertools/kafka avro-js
对于Protobuf计划:
npm install @aws-lambda-powertools/kafka protobufjs
此外,如果想要使用输出解析 标准Schema,您可以安装 任何支持的库例如: Zod, Valibot, 或 ArkType.
取消序列化
卡夫卡消费者公用设施将生的卡夫卡事件转变为直观的处理格式. 要有效处理信件,您需要配置一个与您的数据格式相匹配的策略.
杰森·斯克马
import { SchemaType, kafkaConsumer } from '@aws-lambda-powertools/kafka';
import type { SchemaConfig } from '@aws-lambda-powertools/kafka/types';
import { Logger } from '@aws-lambda-powertools/logger';
const logger = new Logger({ serviceName: 'kafka-consumer' });
const schemaConfig = {
value: {
type: SchemaType.JSON,
},
} satisfies SchemaConfig;
export const handler = kafkaConsumer(async (event, _context) => {
for (const { value } of event.records) {
logger.info('received value', { value });
}
}, schemaConfig);
阿夫罗·斯克马
import { readFileSync } from 'node:fs';
import { SchemaType, kafkaConsumer } from '@aws-lambda-powertools/kafka';
import type { SchemaConfig } from '@aws-lambda-powertools/kafka/types';
import { Logger } from '@aws-lambda-powertools/logger';
const logger = new Logger({ serviceName: 'kafka-consumer' });
const schemaConfig = {
value: {
type: SchemaType.AVRO,
schema: readFileSync(new URL('./user.avsc', import.meta.url), 'utf8'),
},
} satisfies SchemaConfig;
export const handler = kafkaConsumer(async (event, _context) => {
for (const { value } of event.records) {
logger.info('received value', { value });
}
}, schemaConfig);
普罗托布夫·舍马
import { SchemaType, kafkaConsumer } from '@aws-lambda-powertools/kafka';
import type { SchemaConfig } from '@aws-lambda-powertools/kafka/types';
import { Logger } from '@aws-lambda-powertools/logger';
import { User } from './samples/user.es6.generated.js'; // protobuf generated class
const logger = new Logger({ serviceName: 'kafka-consumer' });
const schemaConfig = {
value: {
type: SchemaType.PROTOBUF,
schema: User,
},
} satisfies SchemaConfig;
export const handler = kafkaConsumer(async (event, _context) => {
for (const { value } of event.records) {
logger.info('received value', { value });
}
}, schemaConfig);
附加解析
您可以使用您的首选解析库解析已解析的数据 。 这可以帮助您将Kafka数据与您的域名计划和应用架构整合,提供类型提示,运行时间解析和验证,以及高级数据转换.
佐德语Name
import { SchemaType, kafkaConsumer } from '@aws-lambda-powertools/kafka';
import type { SchemaConfig } from '@aws-lambda-powertools/kafka/types';
import { Logger } from '@aws-lambda-powertools/logger';
import { z } from 'zod/v4';
const logger = new Logger({ serviceName: 'kafka-consumer' });
const OrderItemSchema = z.object({
productId: z.string(),
quantity: z.number().int().positive(),
price: z.number().positive(),
});
const OrderSchema = z.object({
id: z.string(),
customerId: z.string(),
items: z.array(OrderItemSchema).min(1, 'Order must have at least one item'),
createdAt: z.iso.datetime(),
totalAmount: z.number().positive(),
});
const schemaConfig = {
value: {
type: SchemaType.JSON,
parserSchema: OrderSchema,
},
} satisfies SchemaConfig;
export const handler = kafkaConsumer<unknown, z.infer<typeof OrderSchema>>(
async (event, _context) => {
for (const record of event.records) {
const {
value: { id, items },
} = record;
logger.setCorrelationId(id);
logger.debug(`order includes ${items.length} items`);
}
},
schemaConfig
);
维利博特
import { SchemaType, kafkaConsumer } from '@aws-lambda-powertools/kafka';
import type { SchemaConfig } from '@aws-lambda-powertools/kafka/types';
import { Logger } from '@aws-lambda-powertools/logger';
import * as v from 'valibot';
const logger = new Logger({ serviceName: 'kafka-consumer' });
const OrderItemSchema = v.object({
productId: v.string(),
quantity: v.pipe(v.number(), v.integer(), v.toMinValue(1)),
price: v.pipe(v.number(), v.integer()),
});
const OrderSchema = v.object({
id: v.string(),
customerId: v.string(),
items: v.pipe(
v.array(OrderItemSchema),
v.minLength(1, 'Order must have at least one item')
),
createdAt: v.pipe(v.string(), v.isoDateTime()),
totalAmount: v.pipe(v.number(), v.toMinValue(0)),
});
const schemaConfig = {
value: {
type: SchemaType.JSON,
parserSchema: OrderSchema,
},
} satisfies SchemaConfig;
export const handler = kafkaConsumer<unknown, v.InferInput<typeof OrderSchema>>(
async (event, _context) => {
for (const record of event.records) {
const {
value: { id, items },
} = record;
logger.setCorrelationId(id);
logger.debug(`order includes ${items.length} items`);
}
},
schemaConfig
);
ArkType 密码
import { SchemaType, kafkaConsumer } from '@aws-lambda-powertools/kafka';
import type { SchemaConfig } from '@aws-lambda-powertools/kafka/types';
import { Logger } from '@aws-lambda-powertools/logger';
import { type } from 'arktype';
const logger = new Logger({ serviceName: 'kafka-consumer' });
const OrderItemSchema = type({
productId: 'string',
quantity: 'number.integer >= 1',
price: 'number.integer',
});
const OrderSchema = type({
id: 'string',
customerId: 'string',
items: OrderItemSchema.array().moreThanLength(0),
createdAt: 'string.date',
totalAmount: 'number.integer >= 0',
});
const schemaConfig = {
value: {
type: SchemaType.JSON,
parserSchema: OrderSchema,
},
} satisfies SchemaConfig;
export const handler = kafkaConsumer<unknown, typeof OrderSchema.infer>(
async (event, _context) => {
for (const record of event.records) {
const {
value: { id, items },
} = record;
logger.setCorrelationId(id);
logger.debug(`order includes ${items.length} items`);
}
},
schemaConfig
);
见 文档 关于如何使用卡夫卡工具的更详细信息.
贡献
如果你有兴趣为这个项目捐款,请参见我们 贡献准则。 。 。 .
路线图
AWS Lambda(英语:TypeScript)的"Powertools"行进图受顾客需求所驱动.
帮助我们确定未来功能或公用事业的优先次序 更新现有 RFC 和功能请求,或 创建新的,在此 GitHub 仓库中.
连接
- 关于 GitHub 讨论的 AWS Lambda 工具数字 : 加入对话
- 电子邮件数字 : aws-lambda-powertools- [email protected] 互联网档案馆的存檔,存档日期2014-12-21
如何支持 AWS Lambda (TypeScript) 的 Powertools ?
成为参考客户
了解哪些公司正在使用这个图书馆对于帮助在内部确定项目的优先次序很重要。 如果您的公司正在使用 Powertools 为 AWS Lambda (TypeScript) , 您可以要求您在 README 文件中添加您的姓名和标志 AWS Lambda( TypeScript) 的辅助电源工具(成为参考) 问题.
除其他外,下列公司使用Powertools:
- Alma 媒体
- 应用自己
- 贝利·纳尔逊
- Banx 软件
- 凯伦特
- 可认证
- 代码
- 埃尔瓦
- 飞行重量
- 欺诈
- 全球数据网
- 关节
- 半径
- 内向
- 本地软件包
- 我们的隐私
- 完美邮递
- 传感器
- 铁丝网股份有限公司
- 绊脚石10
- 学校
分享你的工作
分享您为 AWS Lambda (TypeScript) 所做的 Powertools 。 看看社区有什么 已经共享 关于AWS Lambda的Powertools (TypeScript).
使用 Lambda 层
这帮助我们了解谁用Powertools为AWS Lambda(英语:TypeScript (TypeScript))以非侵入性的方式使用Powertools,并帮助我们获得未来对其他AWS Lambda语言的Powertool的投资. 何时 使用图层,您可以添加Powertools作为dev的依赖性,以不影响开发过程.
许可证
这个图书馆根据MIT-0许可证获得许可. 见LICENSE文件.