如何在iOS应用程序中实现亚马逊动态流实时处理数据

Question

我已经读过了(AWS)的动态流(实时数据处理)的文档。但是，我对如何实现Kinesis感到困惑。

• 在项目中为iOS安装了SDK。
• 带有区域和池ID的认知凭证的书面代码。
• 成功地用AWSServiceConfiguration完成了。
• 在视图控制器中，我编写了一个代码来保存AWSKinesisRecorder流中的数据。

但是我怎么知道数据正在成功地保存在流中呢？我怎样才能把打印日志放进去？

我想在控制台中显示成功保存的数据。

AppDelegate Code:


 func application(_ application: UIApplication, didFinishLaunchingWithOptions launchOptions: [UIApplicationLaunchOptionsKey: Any]?) -> Bool {
        AWSLogger.default().logLevel = .verbose
        
        let CognitoPoolID = "ap-northeast-1:430840dc-4df5-469f"
        let Region = AWSRegionType.APNortheast1
        let credentialsProvider = AWSCognitoCredentialsProvider(regionType:Region,identityPoolId:CognitoPoolID)
        
        let configuration = AWSServiceConfiguration(region:Region, credentialsProvider:credentialsProvider)
        AWSServiceManager.default().defaultServiceConfiguration = configuration
        
        return true
    }

ViewController代码：

import UIKit 
import AWSKinesis

class ViewController: UIViewController {

       var kinesisRecorder : AWSKinesisRecorder! = nil

       override func viewDidLoad() {
          super.viewDidLoad()
    
        //use AWSKinesisRecorder with Amazon Kinesis. The following snippet returns a shared instance of the Amazon Kinesis service client
         kinesisRecorder = AWSKinesisRecorder.default()
    
    
          configureAWSKinesisRecorder()
        saveDataInStream()
    }

    override func didReceiveMemoryWarning() {
        super.didReceiveMemoryWarning()
    }

    /*
     * Method to configure perties of AWSKinesisRecorder
     */
    func configureAWSKinesisRecorder()  {
        //The diskAgeLimit property sets the expiration for cached requests. When a request exceeds the limit, it's discarded. The default is no age limit.

        kinesisRecorder.diskAgeLimit = TimeInterval(30 * 24 * 60 * 60); // 30 days
    
    
        //The diskByteLimit property holds the limit of the disk cache size in bytes. If the storage limit is exceeded, older requests are discarded. The default value is 5 MB. Setting the value to 0 means that there's no practical limit.

        kinesisRecorder.diskByteLimit = UInt(10 * 1024 * 1024); // 10MB
    
    
        //The notficationByteThreshold property sets the point beyond which Kinesis issues a notification that the byte threshold has been reached. The default value is 0, meaning that by default Amazon Kinesis doesn't post the notification.

        kinesisRecorder.notificationByteThreshold = UInt(5 * 1024 * 1024); // 5MB
    
    }

    /*
     * Method to save real time data in AWS stream
     */
    func saveDataInStream()  {
   
    
        //Both saveRecord and submitAllRecords are asynchronous operations, so you should ensure that saveRecord is complete before you invoke submitAllRecords.
    
        // Create an array to store a batch of objects.
        var tasks = Array<AWSTask<AnyObject>>()
        for i in 0...5 {
            //create an NSData object .
            //StreamName should be a string corresponding to the name of your Kinesis stream.
            //save it locally in kinesisRecorder instances

            tasks.append(kinesisRecorder!.saveRecord(String(format: "TestString-%02d", i).data(using: .utf8), streamName: "my_stream")!)
        }
    
        ////submitAllRecords sends all locally saved requests to the Amazon Kinesis service.
        AWSTask(forCompletionOfAllTasks: tasks).continueOnSuccessWith(block: { (task:AWSTask<AnyObject>) -> AWSTask<AnyObject>? in
            return self.kinesisRecorder?.submitAllRecords()
        }).continueWith(block: { (task:AWSTask<AnyObject>) -> Any? in
            if let error = task.error as? NSError {
                print("Error: \(error)")
            }
            return nil
        }
    
        )
    
    
    
    } 
}

Answer 1

您可以直接使用AWSKinesis而不是使用AWSKinesisRecorder。

putRecords方法(http://docs.aws.amazon.com/AWSiOSSDK/latest/Classes/AWSKinesis.html#//api/name/putRecords:completionHandler:)将显式地将记录写入流，而不是将记录抽象到记录器中。

该方法为您提供了一个completionHandler块，您可以使用它来准确地查看Kinesis响应的内容，无论是成功的记录it还是错误代码。从那里，您可以登录到您的控制台或做任何您喜欢的响应。请注意，对于对putRecords的调用，需要手动对记录进行批处理。您可以使用用户默认值或领域在本地存储事件，然后按计划将它们刷新到Kinesis。这是录音机免费给你的东西。

或者，如果您只想看到事件正在成功地通过您的Kinesis流，您可以将一个Kinesis流连接到您的Kinesis流，您可以配置该流将您的事件数据传递给S3、ElasticSearch或Amazon：https://aws.amazon.com/kinesis/data-firehose/。

Answer 2

在Kinesis中没有任何功能可以执行您希望它做的事情。但是，另一种方法是查询流将数据放入的服务(S3、dynamo等)

Answer 3

另一种方法是连接该流的使用者。例: KinesisFirehose。然后，从消防软管，您可以将其输出到一个S3桶。