{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/15102","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/15102/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/15102/comments","events_url":"https://api.github.com/repos/apache/spark/issues/15102/events","html_url":"https://github.com/apache/spark/pull/15102","id":177025647,"number":15102,"title":"[SPARK-17346][SQL] Add Kafka source for Structured Streaming","user":{"login":"zsxwing","id":1000778,"avatar_url":"https://avatars.githubusercontent.com/u/1000778?v=3","gravatar_id":"","url":"https://api.github.com/users/zsxwing","html_url":"https://github.com/zsxwing","followers_url":"https://api.github.com/users/zsxwing/followers","following_url":"https://api.github.com/users/zsxwing/following{/other_user}","gists_url":"https://api.github.com/users/zsxwing/gists{/gist_id}","starred_url":"https://api.github.com/users/zsxwing/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/zsxwing/subscriptions","organizations_url":"https://api.github.com/users/zsxwing/orgs","repos_url":"https://api.github.com/users/zsxwing/repos","events_url":"https://api.github.com/users/zsxwing/events{/privacy}","received_events_url":"https://api.github.com/users/zsxwing/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":7,"created_at":"2016-09-14T21:25:38Z","updated_at":"2016-09-16T00:08:55Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/15102","html_url":"https://github.com/apache/spark/pull/15102","diff_url":"https://github.com/apache/spark/pull/15102.diff","patch_url":"https://github.com/apache/spark/pull/15102.patch"},"body":"## What changes were proposed in this pull request?\r\n\r\nThis PR adds a new project ` external/kafka-0-10-sql` for Structured Streaming Kafka source.\r\n\r\nIt's based on the design doc: https://docs.google.com/document/d/19t2rWe51x7tq2e5AOfrsM9qb8_m7BRuv9fel9i0PqR8/edit?usp=sharing\r\n\r\n@tdas did most of work and part of them was inspired by @koeninger's work.\r\n\r\n## How was this patch tested?\r\n\r\nThe new unit tests.\r\n"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/247488433","html_url":"https://github.com/apache/spark/pull/15102#issuecomment-247488433","issue_url":"https://api.github.com/repos/apache/spark/issues/15102","id":247488433,"user":{"login":"marmbrus","id":527,"avatar_url":"https://avatars.githubusercontent.com/u/527?v=3","gravatar_id":"","url":"https://api.github.com/users/marmbrus","html_url":"https://github.com/marmbrus","followers_url":"https://api.github.com/users/marmbrus/followers","following_url":"https://api.github.com/users/marmbrus/following{/other_user}","gists_url":"https://api.github.com/users/marmbrus/gists{/gist_id}","starred_url":"https://api.github.com/users/marmbrus/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/marmbrus/subscriptions","organizations_url":"https://api.github.com/users/marmbrus/orgs","repos_url":"https://api.github.com/users/marmbrus/repos","events_url":"https://api.github.com/users/marmbrus/events{/privacy}","received_events_url":"https://api.github.com/users/marmbrus/received_events","type":"User","site_admin":false},"created_at":"2016-09-16T00:08:54Z","updated_at":"2016-09-16T00:08:54Z","body":"> This already does depend on most of the existing Kafka DStream implementation....\r\n\r\nI pushed for this code to be copied rather than refactored because I think this is the right direction long term.  While it is nice to minimize inter-project dependencies, that is not really the motivation.  While the code is very similar now, there a bunch of things I'd like to start changing:\r\n - I don't think that all the classes need to be type parameterized.  Our interface  SQL has its own type system, analyser, and interface to the type system of the JVM (encoders).  We should be using that.  Operators in SQL do not type parameterize in general.\r\n - To optimize performance, there are several tricks we might want to play eventually (maybe prefetching data during execution, etc).\r\n\r\nThese are just ideas, but given that DStreams and Structured Streaming have significantly different models and user interfaces, I don't think that we want to tie ourselves to the same internals.  If we identify utilities that are needed by both, then we should pull those out and share them.\r\n\r\n> Users are going to change topicpartitions whether you want them to or not...\r\n\r\nI think we need to take a step back here and look at what is actually required by the `Offset` interface.  We don't need to handle the general problem of is kafka `Offset A` from `Topic 1` before or after kafka `Offset B` from `Topic 2`.  We'll only ever look at full `KafkaOffsets` that represent slices in time across a couple of topicpartitions and were returned by the `Source` (possibly with other annotations the Source wants to add).  The only questions that the the `StreamExecution` will ask today are:\r\n - Does `x: KafkaOffset == y: KafkaOffset` (i.e. is there new data since the last time I checked)?\r\n - Give me the data between `KafkaOffset` x and `KafkaOffset` y for all included topicpartitions.\r\n\r\nThe final version of this `Source` should almost certainly support wildcards with topicpartitions that change on the fly.  Since it seems this is one of the harder problems to solve.  As a strawman, I'd propose that we only support static lists of topics in this PR and possibly even static partitions.  I want to get users to kick the tires on structured streaming in general and report whats missing so we can all prioritize our engineering effort.\r\n\r\n> It's clear from the jira this shouldn't get rushed into 2.0.1, let's do this as right as possible given the circumstances.\r\n\r\nAgreed, if this doesn't make 2.0.1 thats fine with me.\r\n\r\n> How can we collaborate on a shared branch? You guys manually copying stuff from my fork doesn't make any sense.\r\n\r\nI typically open PRs against [the PR author's branch](https://github.com/zsxwing/spark/tree/kafka-source) when I want to collaborate more directly.\r\n\r\n> Can you give some specific technical direction as to how users can communicate the type for key and value, without having to map over the stream as is done in this PR?\r\n\r\nI'd like this to work the same as other calculations in SQL, using column expressions.\r\n\r\n```scala\r\ndf.withColumn(\"value\", $\"value\".cast(\"string\"))\r\n```\r\n\r\nI'd also like to create expressions that work well with kafka specific deserializers, as well as integrations with our json parsing and maybe Avro too.  The nice thing about this path is it fits with other SQL APIs and should play nicely with Dataset `Encoder`s, UDFs, our expression library, etc.\r\n\r\nDoes that seem reasonable?  Is this missing important cases?\r\n\r\n"}}
{"actor":{"display_login":"marmbrus"}}