{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/7825","labels_url":"https://api.github.com/repos/apache/spark/issues/7825/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/7825/comments","events_url":"https://api.github.com/repos/apache/spark/issues/7825/events","html_url":"https://github.com/apache/spark/pull/7825","id":98353944,"number":7825,"title":"[SPARK-9217][STREAMING] Make the kinesis receiver reliable by recording sequence numbers","user":{"login":"tdas","id":663212,"avatar_url":"https://avatars.githubusercontent.com/u/663212?v=3","gravatar_id":"","url":"https://api.github.com/users/tdas","html_url":"https://github.com/tdas","followers_url":"https://api.github.com/users/tdas/followers","following_url":"https://api.github.com/users/tdas/following{/other_user}","gists_url":"https://api.github.com/users/tdas/gists{/gist_id}","starred_url":"https://api.github.com/users/tdas/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tdas/subscriptions","organizations_url":"https://api.github.com/users/tdas/orgs","repos_url":"https://api.github.com/users/tdas/repos","events_url":"https://api.github.com/users/tdas/events{/privacy}","received_events_url":"https://api.github.com/users/tdas/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":56,"created_at":"2015-07-31T09:41:51Z","updated_at":"2015-08-04T21:41:44Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/7825","html_url":"https://github.com/apache/spark/pull/7825","diff_url":"https://github.com/apache/spark/pull/7825.diff","patch_url":"https://github.com/apache/spark/pull/7825.patch"},"body":"This PR is the second one in the larger issue of making the Kinesis integration reliable and provide WAL-free at-least once guarantee. It is based on the design doc - https://docs.google.com/document/d/1k0dl270EnK7uExrsCE7jYw7PYx0YC935uBcxn3p0f58/edit\r\n\r\nIn this PR, I have updated the Kinesis Receiver to do the following. \r\n- Control the block generation, by creating its own BlockGenerator with own callback methods and using it to keep track of the ranges of sequence numbers that go into each block.\r\n- More specifically, as the KinesisRecordProcessor provides small batches of records, the records are atomically inserted into the block (that is, either the whole batch is in the block, or not). Accordingly the sequence number range of the batch is recorded. Since there may be many batches added to a block, the receiver tracks all the range of sequence numbers that is added to a block.\r\n- When the block is ready to be pushed, the block is pushed and the ranges are reported as metadata of the block. In addition, the ranges are used to find out the latest sequence number for each shard that can be checkpointed through the DynamoDB.\r\n- Periodically, each KinesisRecordProcessor checkpoints the latest successfully stored sequence number for it own shard.\r\n- The array of ranges in the block metadata is used to create KinesisBackedBlockRDDs. The ReceiverInputDStream has been slightly refactored to allow the creation of KinesisBackedBlockRDDs instead of the WALBackedBlockRDDs. \r\n\r\nThings to be done\r\n- [x] Add new test to verify that the sequence numbers are recovered.\r\n\r\n"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/127770743","html_url":"https://github.com/apache/spark/pull/7825#issuecomment-127770743","issue_url":"https://api.github.com/repos/apache/spark/issues/7825","id":127770743,"user":{"login":"SparkQA","id":8108735,"avatar_url":"https://avatars.githubusercontent.com/u/8108735?v=3","gravatar_id":"","url":"https://api.github.com/users/SparkQA","html_url":"https://github.com/SparkQA","followers_url":"https://api.github.com/users/SparkQA/followers","following_url":"https://api.github.com/users/SparkQA/following{/other_user}","gists_url":"https://api.github.com/users/SparkQA/gists{/gist_id}","starred_url":"https://api.github.com/users/SparkQA/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/SparkQA/subscriptions","organizations_url":"https://api.github.com/users/SparkQA/orgs","repos_url":"https://api.github.com/users/SparkQA/repos","events_url":"https://api.github.com/users/SparkQA/events{/privacy}","received_events_url":"https://api.github.com/users/SparkQA/received_events","type":"User","site_admin":false},"created_at":"2015-08-04T21:41:44Z","updated_at":"2015-08-04T21:41:44Z","body":"  [Test build #39755 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/39755/console) for   PR 7825 at commit [`bf31e22`](https://github.com/apache/spark/commit/bf31e220eca0016b776bca6ad55be624c71969db).\n * This patch **fails Scala style tests**.\n * This patch merges cleanly.\n * This patch adds the following public classes _(experimental)_:\n  * `case class SequenceNumberRanges(ranges: Seq[SequenceNumberRange]) `\n  * `      ParamDesc[Array[Double]](\"thresholds\", \"Thresholds in multi-class classification\" +`\n  * `  final val thresholds: DoubleArrayParam = new DoubleArrayParam(this, \"thresholds\", \"Thresholds in multi-class classification to adjust the probability of predicting each class. Array must have length equal to the number of classes, with values >= 0. The class with largest value p/t is predicted, where p is the original probability of that class and t is the class' threshold.\", (t: Array[Double]) => t.forall(_ >= 0))`\n"}}