如何将数据从服务器放入 Kinesis Stream [英] How to put data from server to Kinesis Stream
问题描述
我是 Kinesis 的新手.阅读我发现的文档，我可以创建 Kinesis Stream 以从 Producer 获取数据.然后使用 KCL 将从 Stream 中读取此数据以进行进一步处理.我了解如何通过实现 IRecordProcessor 来编写 KCL 应用程序.
I am new to Kinesis. Reading out the documentation i found i can create the Kinesis Stream to get data from Producer. Then using KCL will read this data from Stream to further processing. I understand how to write the KCL application by implemeting IRecordProcessor .
然而，如何将数据放在 Kinesis 流上的第一阶段对我来说仍然不清楚.我们是否有一些需要实现的 AWS API.
However the very first stage as how to put data on Kinesis stream is still not clear to me. Do we have some AWS API which does need implementation to achieve this.
场景:我有一个服务器，它不断从文件夹中的各种来源获取数据.每个文件夹都包含文本文件，其行包含进一步分析工作所需的属性.我必须将所有这些数据推送到 Kinesis Stream.
Scenarios: I have an server which is contineously getting data from various sources in the folders. Each folder is containing the text file whose rows are containing the required attributes for furhter analytical work. i have to push all these data to Kinesis Stream.
我需要在下面的类 putData 方法中使用下面的代码在 Kinesis 流中输出
I need code something as below below class putData method wil be used to out in Kinesis stream
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 流上的记录.我是否需要无限循环并读取所有文件，然后执行此操作或某些框架，以便在容错、单点故障的情况下更好地执行此操作.任何帮助将不胜感激.
However reading file from the source folder from the server and then what design i should use to call putData to get the record on Kinesis stream. Do i need infinite loop and reading all files and then do this or some framework which will better do this with care of fault tolerance , single point of failure all . Any help would be greatly appreciated.
简而言之:我需要一种更好的技术来将定期生成的数据放入 Kinesis Stream 数据是定期生成到服务器的.谢谢
Briefly: I need a better technique to put regularly generated data to Kinesis Stream the data is generated at regular interval to server. Thanks
推荐答案
So it seems you are already using... http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/kinesis/AmazonKinesisClient.html
你想要的具体方法如下.
Specific method you want is as follows.
您需要流名称、记录和流密钥.http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/kinesis/model/PutRecordResult.html
You need a stream name, record, and stream key. http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/kinesis/model/PutRecordResult.html
但你似乎拥有这一切?
然后你需要一个程序运行总是拖尾你的服务器日志文件，当有一个新行时它会推送它.
You would then need a program running always tailing your server log file and when ever there is a new line it will push this.
但您的数据只会保留 24 小时.然后，您需要一个工作程序来使用数据并将其放置在其他一些 AWS 资源中.
But your data will only sit their for 24 hours. You then need a worker program to consume the data and place it in some other AWS resource.
这篇关于如何将数据从服务器放入 Kinesis Stream的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
