Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/13342#discussion_r64957648
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/ForeachWriter.scala
---
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql
+
+import org.apache.spark.annotation.Experimental
+
+/**
+ * :: Experimental ::
+ * A writer to consume data generated by a [[ContinuousQuery]].
+ *
+ * @since 2.0.0
+ */
+@Experimental
+trait ForeachWriter[T] extends Serializable {
--- End diff --
I agree that we can't ourselves guarantee exactly once here, but we should
at least give the user the tools to implement that themselves. Kafka has some
good thoughts
[here](https://cwiki.apache.org/confluence/display/KAFKA/Idempotent+Producer).
In particular I think we are missing several pieces of data:
- the partition
- some id for the record (unless we are promising they always come in the
same order? and they are responsible for counting on their own.)
It also seems by having open return a boolean we don't have a great
mechanism to handle partial failures. An alternative proposal might be:
```scala
trait ForeachWriter[T] {
def open(pid: Long): Unit
def process(recId: Long, value: T): Unit
def close(error: Throwable): Unit
}
```
- `pid` can be a combination of the `partitionId` and the `batchId`,
though in the future we could change it to partition + checkpoint id. The only
promise here is that once we call `close` where `error != null` you have seen
all of the tuples that you will ever see for a given `pid` and if we repeat,
that set will not change. This lets you do de-duplication without remembering
an id for every record ever seen forever.
- `recId` needs to be consistent (i.e. always be the same tuple) for any
given `pid`. For now we can enforce this by sorting each partition on all
columns, but there are certainly other faster ways we can explore in the future.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
