根据AWS Kinesis Firehose的文档,该功能已经被废弃,无法直接通过控制台进行演示数据测试。不过,您可以使用AWS Lambda函数和CloudWatch日志组来测试数据。以下是一个简单的Python Lambda函数示例,可以将数据写入CloudWatch日志组:
import json
import boto3
import logging
client = boto3.client('logs')
logger = logging.getLogger()
logger.setLevel(logging.INFO)
def lambda_handler(event, context):
logger.info('Received event: ' + json.dumps(event))
logger.info('DeliveryStreamArn: ' + event['deliveryStreamArn'])
logger.info('DeliveryStreamName: ' + event['deliveryStreamName'])
for record in event['records']:
payload = base64.b64decode(record['data']).decode('utf-8')
logger.info('Payload: ' + payload)
response = client.create_log_group(
logGroupName='firehose-test',
tags={
'Source': 'FirehoseTest'
}
)
response = client.create_log_stream(
logGroupName='firehose-test',
logStreamName='test-stream'
)
response = client.put_log_events(
logGroupName='firehose-test',
logStreamName='test-stream',
logEvents=[
{
'timestamp': int(round(time.time() * 1000)),
'message': payload
}
]
)
logger.info('PutLogEvents response: ' + json.dumps(response))
return 'Successfully processed {} records.'.format(len(event['records']))
您可以将此Lambda函数与Kinesis Firehose流程相关联,以测试数据并将其写入CloudWatch日志组。请注意,您需要在CloudWatch日志组中手动创建日志组和日志流。此外,您需要在Lambda函数的IAM角色中授权一些相关权限,使其能够访问CloudWatch和Kinesis Firehose。
此外,您还可以编写其他脚本或小型应用来模拟和测试数据,然后