如何将数据从服务器放入 Kinesis Stream答案

【问题标题】：How to put data from server to Kinesis Stream如何将数据从服务器放入 Kinesis Stream
【发布时间】：2014-06-21 16:12:09
【问题描述】：

我是 Kinesis 的新手。阅读我发现的文档，我可以创建 Kinesis Stream 以从 Producer 获取数据。然后使用 KCL 将从 Stream 中读取此数据以进行进一步处理。我了解如何通过实现 IRecordProcessor 来编写 KCL 应用程序。

但是，我仍然不清楚如何将数据放入 Kinesis 流的第一阶段。我们是否有一些 AWS API 需要实施才能实现这一点。

场景：我有一台服务器，它不断地从文件夹中的各种来源获取数据。每个文件夹都包含文本文件，其行包含进一步分析工作所需的属性。我必须将所有这些数据推送到 Kinesis Stream。

我需要下面的代码，类 putData 方法将用于在 Kinesis 流中输出

public class Put {

    AmazonKinesisClient kinesisClient;

    Put()
    {
        String accessKey = "My Access Key here" ;
        String secretKey = "My Secret Key here" ;
        AWSCredentials credentials = new BasicAWSCredentials(accessKey, secretKey);
       kinesisClient = new AmazonKinesisClient(credentials);
       kinesisClient.setEndpoint("kinesis.us-east-1.amazonaws.com", "kinesis", "us-east-1");
        System.out.println("starting the Put Application");
    }

    public void putData(String fileContent,String session) throws Exception
    {
         final String myStreamName = "ClickStream";

            PutRecordRequest putRecordRequest = new PutRecordRequest();
            putRecordRequest.setStreamName(myStreamName);
            String putData = fileContent;
            putRecordRequest.setData(ByteBuffer.wrap(putData.getBytes()));
            putRecordRequest.setPartitionKey("session"+session);
            PutRecordResult putRecordResult = kinesisClient.putRecord(putRecordRequest);
            System.out.println("Successfully putrecord, partition key : " + putRecordRequest.getPartitionKey()
                    + ", ShardID : " + putRecordResult.getShardId());
            System.out.println(fileContent);
            System.out.println("Sequence Number: "+putRecordResult.getSequenceNumber());

            System.out.println("Data has been PUT successfully");


    }
}

但是从服务器的源文件夹中读取文件，然后我应该使用什么设计来调用 putData 以获取 Kinesis 流上的记录。我是否需要无限循环并读取所有文件，然后执行此操作或某些框架会更好地执行此操作，同时兼顾容错性和单点故障。任何帮助将不胜感激。

简而言之：我需要一种更好的技术来将定期生成的数据放到 Kinesis Stream 中，这些数据会定期生成到服务器。谢谢

【问题讨论】：

标签： java amazon-web-services rabbitmq amazon-kinesis

【解决方案1】：

如果您要拖尾某些文件，请尝试使用 Fluentd。 http://www.fluentd.org/

Amazon Kinesis 有一个非常好的插件。 https://github.com/awslabs/aws-fluent-plugin-kinesis

【讨论】：

【解决方案2】：

看来您已经在使用... http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/kinesis/AmazonKinesisClient.html

你想要的具体方法如下。

您需要一个流名称、记录和流密钥。 http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/kinesis/model/PutRecordResult.html

但你似乎拥有了这一切？

然后，您将需要一个始终跟踪您的服务器日志文件的程序运行，并且当有新行时它会推送它。

但您的数据只能保留 24 小时。然后，您需要一个工作程序来使用数据并将其放置在其他一些 AWS 资源中。

【讨论】：

是的，我正在将数据从 Kinesis 移动到 S3。我一直在寻找一些现成的解决方案，以每天从我的服务器中读取文件夹中的文件，并将所有这些数据放入 Kinesis 流。那么在我的服务器中，我有多个不同日期的文件夹，并且每天都包含许多带有日志信息的文件。我想将此传输到 Kinesis 流。在这个级别上，我认为我可以编写带有一些线程延迟的无限循环的简单程序，以继续读取事件并在不存在某些已经证明的解决方案时转移到 Kinesis。谢谢
我可以使用 RabbitMQ 将数据放入 Kinesis Stream。 ?
亚马逊不提供任何开箱即用的推送程序。您必须自己创建它。不熟悉 RabbitMQ
WatchService API 的 Java7 有趣特性非常适合处理我的文件问题。现在我可以在创建\更改时读取文件等。但是，您能否向我推荐一些选择分区键的好技巧。 ?
想象一副纸牌，一个好的分区键是颜色，但更好的分区键是花色。这取决于您的数据，但我使用了诸如 EventType、UserID、Country+StateofRecord 之类的东西。您想尝试在读者之间平均分配数据。

【解决方案3】：

您可以使用 Amazon kinesis 代理来监控一组文件，它们可以将数据流式传输到 kinesis。

http://docs.aws.amazon.com/streams/latest/dev/writing-with-agents.html

【讨论】：

【解决方案4】：

如果您尝试提取日志文件，请尝试使用 Fluentd。 Fluentd可以连续tail log文件，做数据缓冲、加密、压缩、重试。

http://docs.fluentd.org/articles/kinesis-stream

Fluentd 的 Kinesis 插件由 Amazon Web Services 自己开发。

【讨论】：