[Docs] Add unbounded source guidance to the Python I/O connector guide

website/www/site/content/en/documentation/io/developing-io-python.md

@@ -69,6 +69,12 @@ for Beam's transform style guidance.
 
 To create a new data source for your pipeline, you'll need to provide the format-specific logic that tells the service how to read data from your input source, and how to split your data source into multiple parts so that multiple worker instances can read your data in parallel.
 
+The `Source` interface described here reads a finite, *bounded* data set. To read
+an infinite, *unbounded* (streaming) data set, use a Splittable DoFn instead of a
+`Source` subclass, as described in
+[Reading from an unbounded source](#reading-unbounded). Unlike the Java SDK, the
+Python SDK does not provide a separate `UnboundedSource` class.
+
 Supply the logic for your new source by creating the following classes:
 
   * A subclass of `BoundedSource`. `BoundedSource` is a source that reads a
@@ -202,6 +208,125 @@ demonstrated in the example above. Use a wrapping `PTransform` instead.
 exposing your sources, and walks through how to create a wrapper.
 
 
+### Reading from an unbounded source {#reading-unbounded}
+
+The classes shown above read a finite, *bounded* data set. To read an infinite,
+*unbounded* (streaming) data set, the Python SDK does not provide a separate
+`UnboundedSource` class the way the Java SDK does. Instead, you implement an
+unbounded source as a [Splittable DoFn](/documentation/programming-guide/#splittable-dofns)
+(SDF), which is also the recommended way to build any new source in Python. A
+Splittable DoFn provides the same capabilities that the Java SDK splits across
+`UnboundedSource` and `UnboundedReader`, namely parallel reads, checkpointing for
+failure recovery, and watermark reporting, through a single API that works for
+both bounded and unbounded data.
+
+Mark the `process` method of your `DoFn` with
+`@beam.DoFn.unbounded_per_element()` to tell the runner that it performs an
+unbounded amount of work per input element, so the resulting `PCollection` is
+unbounded:
+
+```py
+from apache_beam.io.watermark_estimators import ManualWatermarkEstimator
+from apache_beam.transforms.window import TimestampedValue
+from apache_beam.utils.timestamp import Duration
+
+
+class _ReadFromMyStream(beam.DoFn):
+  @beam.DoFn.unbounded_per_element()
+  def process(
+      self,
+      element,
+      tracker=beam.DoFn.RestrictionParam(MyStreamRestrictionProvider()),
+      watermark_estimator=beam.DoFn.WatermarkEstimatorParam(
+          ManualWatermarkEstimator.default_provider())):
+    for position, record, event_time in read_stream(
+        element, tracker.current_restriction()):
+      if not tracker.try_claim(position):
+        # The runner took back the rest of the restriction; it resumes later.
+        return
+      watermark_estimator.set_watermark(event_time)
+      yield TimestampedValue(record, event_time)
+    # Out of data for now: checkpoint and resume after a short delay.
+    tracker.defer_remainder(Duration(seconds=5))
+```
+
+The following pieces map the concerns that the Java SDK handles with
+`UnboundedSource` and `UnboundedReader` onto their Splittable DoFn equivalents:
+
+  * **Parallel reads and splitting.** Provide a `RestrictionProvider` (in
+    [apache_beam.transforms.core](https://beam.apache.org/releases/pydoc/{{< param release_latest >}}/apache_beam.transforms.core.html))
+    that describes the work for each element. `initial_restriction` returns the
+    restriction to read, which for an unbounded stream is typically an
+    open-ended range (such as an offset range with no upper bound), and `split`
+    or `split_and_size` produces the initial parallel splits, the counterpart
+    of Java's `UnboundedSource.split`. The runner can subdivide work further at
+    run time by calling `try_split` on the restriction tracker.
+
+  * **Checkpointing and failure recovery.** In the Splittable DoFn model the
+    *residual restriction* is the checkpoint. A residual is produced when the
+    runner splits a restriction that is being processed (after which
+    `try_claim` returns `False`) or when `process` checkpoints itself with
+    `defer_remainder`. The runner durably stores the residual and re-invokes
+    the DoFn from it, so no records are lost or read twice. This replaces
+    Java's `getCheckpointMark` and `getCheckpointMarkCoder`. Supply a coder for
+    your restriction type through `RestrictionProvider.restriction_coder` so
+    the runner can persist the restriction.
+
+  * **Yielding and resuming.** An unbounded reader must yield control when it
+    has temporarily run out of data and be resumed later. Call
+    `tracker.defer_remainder(Duration(seconds=...))` and return from `process`
+    to checkpoint the remaining work and schedule it to resume after the given
+    delay. This is the polling loop that, in Java, lives inside
+    `UnboundedReader.advance`.
+
+  * **Watermarks.** Report a watermark (the approximate lower bound on the event
+    times of future records) so that downstream windowing and triggers can
+    estimate completeness. Attach a `WatermarkEstimatorProvider` through
+    `beam.DoFn.WatermarkEstimatorParam` and advance the watermark as you read.
+    The SDK ships several estimators in
+    [apache_beam.io.watermark_estimators](https://github.com/apache/beam/blob/master/sdks/python/apache_beam/io/watermark_estimators.py):
+    `ManualWatermarkEstimator` (you call `set_watermark` explicitly),
+    `MonotonicWatermarkEstimator`, and `WalltimeWatermarkEstimator`. This
+    replaces Java's `UnboundedReader.getWatermark`.
+
+  * **Deduplication.** Java's `UnboundedSource` exposes `requiresDeduping` and
+    `UnboundedReader.getCurrentRecordId` so the runner can drop duplicate
+    records. The Splittable DoFn model has no per-record ID hook; exactly-once
+    processing of each restriction is handled by the runner. If the upstream
+    system can deliver the same record more than once and you need to remove
+    those duplicates, apply the
+    [Deduplicate or DeduplicatePerKey](https://beam.apache.org/releases/pydoc/{{< param release_latest >}}/apache_beam.transforms.deduplicate.html)
+    transform after the read.
+
+  * **Reading a bounded amount from an unbounded source.** Java caps an
+    unbounded read with `.withMaxNumRecords` or `.withMaxReadTime`. In Python, a
+    Splittable DoFn participates in pipeline *drain* through
+    `RestrictionProvider.truncate`, which converts the remaining unbounded
+    restriction into a finite one (return `None` to drop it). You can also bound
+    the output with ordinary transforms downstream of the read.
+
+#### Unbounded source building blocks
+
+You usually do not need to write an unbounded Splittable DoFn from scratch. The
+SDK provides higher-level, reusable transforms:
+
+  * [`PeriodicImpulse` and `PeriodicSequence`](https://github.com/apache/beam/blob/master/sdks/python/apache_beam/transforms/periodicsequence.py)
+    emit elements on a fixed interval and are useful for driving periodic work
+    and for building simple polling sources.
+
+  * The experimental `Watch` transform
+    ([watch.py](https://github.com/apache/beam/blob/master/sdks/python/apache_beam/io/watch.py))
+    continuously watches a growing set of outputs for each input element,
+    updating the watermark and terminating per a user-supplied condition. It is
+    the engine behind periodic file discovery and other polling sources, and is
+    the Python counterpart of Java's `Watch.growthOf`.
+
+For the full Splittable DoFn API and lifecycle, see the
+[Splittable DoFns](/documentation/programming-guide/#splittable-dofns) section of
+the programming guide and the
+[new I/O connector overview](/documentation/io/developing-io-overview/).
+
+
 ## Using the FileBasedSink abstraction
 
 If your data source uses files, you can implement the [FileBasedSink](https://beam.apache.org/releases/pydoc/{{< param release_latest >}}/apache_beam.io.filebasedsink.html)