com.hazelcast.jet.pipeline

## Class Sources

- java.lang.Object
- - com.hazelcast.jet.pipeline.Sources

-     public final class Sources
      extends Object

  Contains factory methods for various types of pipeline sources. To start building a pipeline, pass a source to `Pipeline.drawFrom(BatchSource)` and you will obtain the initial `BatchStage`. You can then attach further stages to it.

  The same pipeline may contain more than one source, each starting its own branch. The branches may be merged with multiple-input transforms such as co-group and hash-join.

  The default local parallelism for sources in this class is 1 or 2, check the documentation of individual methods.

- - ### Method Summary

    All Methods Static Methods Concrete Methods
    Modifier and Type | Method and Description
    --- | ---
    `static <T> BatchSource<T>` | `batchFromProcessor(String sourceName, ProcessorMetaSupplier metaSupplier)` Returns a bounded (batch) source constructed directly from the given Core API processor meta-supplier.
    `static <K,V> BatchSource<Map.Entry<K,V>>` | `cache(String cacheName)` Returns a source that fetches entries from a Hazelcast `ICache` with the given name and emits them as `Map.Entry`.
    `static <T,K,V> StreamSource<T>` | `cacheJournal(String cacheName, DistributedPredicate<? super EventJournalCacheEvent<K,V>> predicateFn, DistributedFunction<? super EventJournalCacheEvent<K,V>,? extends T> projectionFn, JournalInitialPosition initialPos)` Returns a source that will stream the `EventJournalCacheEvent` events of a Hazelcast `ICache` with the specified name.
    `static <K,V> StreamSource<Map.Entry<K,V>>` | `cacheJournal(String cacheName, JournalInitialPosition initialPos)` Convenience for `cacheJournal(String, DistributedPredicate, DistributedFunction, JournalInitialPosition)` which will pass only `CREATED` and `UPDATED` events and will project the event's key and new value into a `Map.Entry`.
    `static BatchSource<String>` | `files(String directory)` A source to read all files in a directory in a batch way.
    `static FileSourceBuilder` | `filesBuilder(String directory)` Returns a builder object that offers a step-by-step fluent API to build a custom source to read files for the Pipeline API.
    `static StreamSource<String>` | `fileWatcher(String watchedDirectory)` A source to stream lines added to files in a directory.
    `static <T> BatchSource<T>` | `jdbc(DistributedSupplier<? extends Connection> connectionSupplier, ToResultSetFunction resultSetFn, DistributedFunction<? super ResultSet,? extends T> createOutputFn)` Returns a source which connects to the specified database using the given `connectionSupplier`, queries the database and creates a result set using the the given `resultSetFn`.
    `static <T> BatchSource<T>` | `jdbc(String connectionURL, String query, DistributedFunction<? super ResultSet,? extends T> createOutputFn)` Convenience for `jdbc(DistributedSupplier, ToResultSetFunction, DistributedFunction)`.
    `static StreamSource<javax.jms.Message>` | `jmsQueue(DistributedSupplier<? extends javax.jms.ConnectionFactory> factorySupplier, String name)` Convenience for `jmsQueueBuilder(DistributedSupplier)`.
    `static JmsSourceBuilder` | `jmsQueueBuilder(DistributedSupplier<? extends javax.jms.ConnectionFactory> factorySupplier)` Returns a builder object that offers a step-by-step fluent API to build a custom JMS `StreamSource` for the Pipeline API.
    `static StreamSource<javax.jms.Message>` | `jmsTopic(DistributedSupplier<? extends javax.jms.ConnectionFactory> factorySupplier, String name)` Convenience for `jmsTopicBuilder(DistributedSupplier)`.
    `static JmsSourceBuilder` | `jmsTopicBuilder(DistributedSupplier<? extends javax.jms.ConnectionFactory> factorySupplier)` Returns a builder object that offers a step-by-step fluent API to build a custom JMS `StreamSource` for the Pipeline API.
    `static <T> BatchSource<T>` | `list(IList<? extends T> list)` Returns a source that emits items retrieved from a Hazelcast `IList`.
    `static <T> BatchSource<T>` | `list(String listName)` Returns a source that emits items retrieved from a Hazelcast `IList`.
    `static <K,V> BatchSource<Map.Entry<K,V>>` | `map(IMap<? extends K,? extends V> map)` Returns a source that fetches entries from the given Hazelcast `IMap` and emits them as `Map.Entry`.
    `static <T,K,V> BatchSource<T>` | `map(IMap<? extends K,? extends V> map, Predicate<? super K,? super V> predicate, DistributedFunction<? super Map.Entry<K,V>,? extends T> projectionFn)` Convenience for `map(IMap, Predicate, Projection)` which uses a `DistributedFunction` as the projection function.
    `static <T,K,V> BatchSource<T>` | `map(IMap<? extends K,? extends V> map, Predicate<? super K,? super V> predicate, Projection<? super Map.Entry<K,V>,? extends T> projection)` Returns a source that fetches entries from the given Hazelcast `IMap`.
    `static <K,V> BatchSource<Map.Entry<K,V>>` | `map(String mapName)` Returns a source that fetches entries from a local Hazelcast `IMap` with the specified name and emits them as `Map.Entry`.
    `static <T,K,V> BatchSource<T>` | `map(String mapName, Predicate<? super K,? super V> predicate, DistributedFunction<? super Map.Entry<K,V>,? extends T> projectionFn)` Convenience for `map(String, Predicate, Projection)` which uses a `DistributedFunction` as the projection function.
    `static <T,K,V> BatchSource<T>` | `map(String mapName, Predicate<? super K,? super V> predicate, Projection<? super Map.Entry<K,V>,? extends T> projection)` Returns a source that fetches entries from a local Hazelcast `IMap` with the specified name.
    `static <T,K,V> StreamSource<T>` | `mapJournal(IMap<? extends K,? extends V> map, DistributedPredicate<? super EventJournalMapEvent<K,V>> predicateFn, DistributedFunction<? super EventJournalMapEvent<K,V>,? extends T> projectionFn, JournalInitialPosition initialPos)` Returns a source that will stream `EventJournalMapEvent`s of the given Hazelcast `IMap`.
    `static <K,V> StreamSource<Map.Entry<K,V>>` | `mapJournal(IMap<? extends K,? extends V> map, JournalInitialPosition initialPos)` Convenience for `mapJournal(IMap, DistributedPredicate, DistributedFunction, JournalInitialPosition)` which will pass only `ADDED` and `UPDATED` events and will project the event's key and new value into a `Map.Entry`.
    `static <T,K,V> StreamSource<T>` | `mapJournal(String mapName, DistributedPredicate<? super EventJournalMapEvent<K,V>> predicateFn, DistributedFunction<? super EventJournalMapEvent<K,V>,? extends T> projectionFn, JournalInitialPosition initialPos)` Returns a source that will stream `EventJournalMapEvent`s of the Hazelcast `IMap` with the specified name.
    `static <K,V> StreamSource<Map.Entry<K,V>>` | `mapJournal(String mapName, JournalInitialPosition initialPos)` Convenience for `mapJournal(String, DistributedPredicate, DistributedFunction, JournalInitialPosition)` which will pass only `ADDED` and `UPDATED` events and will project the event's key and new value into a `Map.Entry`.
    `static <K,V> BatchSource<Map.Entry<K,V>>` | `remoteCache(String cacheName, ClientConfig clientConfig)` Returns a source that fetches entries from the Hazelcast `ICache` with the specified name in a remote cluster identified by the supplied `ClientConfig` and emits them as `Map.Entry`.
    `static <T,K,V> StreamSource<T>` | `remoteCacheJournal(String cacheName, ClientConfig clientConfig, DistributedPredicate<? super EventJournalCacheEvent<K,V>> predicateFn, DistributedFunction<? super EventJournalCacheEvent<K,V>,? extends T> projectionFn, JournalInitialPosition initialPos)` Returns a source that will stream the `EventJournalCacheEvent` events of the Hazelcast `ICache` with the specified name from a remote cluster.
    `static <K,V> StreamSource<Map.Entry<K,V>>` | `remoteCacheJournal(String cacheName, ClientConfig clientConfig, JournalInitialPosition initialPos)` Convenience for `remoteCacheJournal(String, ClientConfig, DistributedPredicate, DistributedFunction, JournalInitialPosition)` which will pass only `CREATED` and `UPDATED` events and will project the event's key and new value into a `Map.Entry`.
    `static <T> BatchSource<T>` | `remoteList(String listName, ClientConfig clientConfig)` Returns a source that emits items retrieved from a Hazelcast `IList` in a remote cluster identified by the supplied `ClientConfig`.
    `static <K,V> BatchSource<Map.Entry<K,V>>` | `remoteMap(String mapName, ClientConfig clientConfig)` Returns a source that fetches entries from the Hazelcast `IMap` with the specified name in a remote cluster identified by the supplied `ClientConfig` and emits them as `Map.Entry`.
    `static <T,K,V> BatchSource<T>` | `remoteMap(String mapName, ClientConfig clientConfig, Predicate<? super K,? super V> predicate, DistributedFunction<? super Map.Entry<K,V>,? extends T> projectionFn)` Convenience for `remoteMap(String, ClientConfig, Predicate, Projection)` which use a `DistributedFunction` as the projection function.
    `static <T,K,V> BatchSource<T>` | `remoteMap(String mapName, ClientConfig clientConfig, Predicate<? super K,? super V> predicate, Projection<? super Map.Entry<K,V>,? extends T> projection)` Returns a source that fetches entries from a remote Hazelcast `IMap` with the specified name in a remote cluster identified by the supplied `ClientConfig`.
    `static <T,K,V> StreamSource<T>` | `remoteMapJournal(String mapName, ClientConfig clientConfig, DistributedPredicate<? super EventJournalMapEvent<K,V>> predicateFn, DistributedFunction<? super EventJournalMapEvent<K,V>,? extends T> projectionFn, JournalInitialPosition initialPos)` Returns a source that will stream the `EventJournalMapEvent` events of the Hazelcast `IMap` with the specified name from a remote cluster.
    `static <K,V> StreamSource<Map.Entry<K,V>>` | `remoteMapJournal(String mapName, ClientConfig clientConfig, JournalInitialPosition initialPos)` Convenience for `remoteMapJournal(String, ClientConfig, DistributedPredicate, DistributedFunction, JournalInitialPosition)` which will pass only `ADDED` and `UPDATED` events and will project the event's key and new value into a `Map.Entry`.
    `static StreamSource<String>` | `socket(String host, int port)` Convenience for `socket(host, port, charset)` with UTF-8 as the charset.
    `static StreamSource<String>` | `socket(String host, int port, Charset charset)` Returns a source which connects to the specified socket and emits lines of text received from it.
    `static <T> StreamSource<T>` | `streamFromProcessor(String sourceName, ProcessorMetaSupplier metaSupplier)` Returns an unbounded (event stream) source constructed directly from the given Core API processor meta-supplier.
    `static <T> StreamSource<T>` | `streamFromProcessorWithWatermarks(String sourceName, Function<WatermarkGenerationParams<? super T>,ProcessorMetaSupplier> metaSupplierFn)` Returns an unbounded (event stream) source that will use the supplied function to create processor meta-suppliers as required by the Core API.

    - ### Methods inherited from class java.lang.Object

      `clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait`

- - ### Method Detail

    - #### batchFromProcessor

          @Nonnull
          public static <T> BatchSource<T> batchFromProcessor(@Nonnull
                                                                       String sourceName,
                                                                       @Nonnull
                                                                       ProcessorMetaSupplier metaSupplier)

      Returns a bounded (batch) source constructed directly from the given Core API processor meta-supplier.
      Parameters:
        `sourceName` - user-friendly source name
        `metaSupplier` - the processor meta-supplier

    - #### streamFromProcessorWithWatermarks

          @Nonnull
          public static <T> StreamSource<T> streamFromProcessorWithWatermarks(@Nonnull
                                                                                       String sourceName,
                                                                                       @Nonnull
                                                                                       Function<WatermarkGenerationParams<? super T>,ProcessorMetaSupplier> metaSupplierFn)

      Returns an unbounded (event stream) source that will use the supplied function to create processor meta-suppliers as required by the Core API. Jet will call the function you supply with the watermark generation parameters and it must return a meta-supplier of processors that will act according to these parameters and emit the watermark items as they specify.

      If you are implementing a custom source processor, be sure to check out the `WatermarkSourceUtil` class that will help you correctly implement watermark item emission.

      Parameters:
        `sourceName` - user-friendly source name
        `metaSupplierFn` - factory of processor meta-suppliers

    - #### streamFromProcessor

          @Nonnull
          public static <T> StreamSource<T> streamFromProcessor(@Nonnull
                                                                         String sourceName,
                                                                         @Nonnull
                                                                         ProcessorMetaSupplier metaSupplier)

      Returns an unbounded (event stream) source constructed directly from the given Core API processor meta-supplier.
      Parameters:
        `sourceName` - user-friendly source name
        `metaSupplier` - the processor meta-supplier

    - #### map

          @Nonnull
          public static <K,V> BatchSource<Map.Entry<K,V>> map(@Nonnull
                                                                       String mapName)

      Returns a source that fetches entries from a local Hazelcast `IMap` with the specified name and emits them as `Map.Entry`. It leverages data locality by making each of the underlying processors fetch only those entries that are stored on the member where it is running.

      The source does not save any state to snapshot. If the job is restarted, it will re-emit all entries.

      If the `IMap` is modified while being read, or if there is a cluster topology change (triggering data migration), the source may miss and/or duplicate some entries.

      The default local parallelism for this processor is 2 (or 1 if just 1 CPU is available).

    - #### map

          @Nonnull
          public static <K,V> BatchSource<Map.Entry<K,V>> map(@Nonnull
                                                                       IMap<? extends K,? extends V> map)

      Returns a source that fetches entries from the given Hazelcast `IMap` and emits them as `Map.Entry`. It leverages data locality by making each of the underlying processors fetch only those entries that are stored on the member where it is running.

      NOTE: Jet only remembers the name of the map you supply and acquires a map with that name on the local cluster. If you supply a map instance from another cluster, no error will be thrown to indicate this.

      The source does not save any state to snapshot. If the job is restarted, it will re-emit all entries.

      If the `IMap` is modified while being read, or if there is a cluster topology change (triggering data migration), the source may miss and/or duplicate some entries.

      The default local parallelism for this processor is 2 (or 1 if just 1 CPU is available).

    - #### map

          @Nonnull
          public static <T,K,V> BatchSource<T> map(@Nonnull
                                                            String mapName,
                                                            @Nonnull
                                                            Predicate<? super K,? super V> predicate,
                                                            @Nonnull
                                                            Projection<? super Map.Entry<K,V>,? extends T> projection)

      Returns a source that fetches entries from a local Hazelcast `IMap` with the specified name. By supplying a `predicate` and `projection` here instead of in separate `map/filter` transforms you allow the source to apply these functions early, before generating any output, with the potential of significantly reducing data traffic. If your data is stored in the IMDG using the portable serialization format, there are additional optimizations available when using `Projections.singleAttribute(java.lang.String)` and `Projections.multiAttribute(java.lang.String...)`) to create your projection instance and using the `GenericPredicates` factory or `PredicateBuilder` to create the predicate. In this case Jet can test the predicate and apply the projection without deserializing the whole object.

      Due to the current limitations in the way Jet reads the map it can't use any indexes on the map. It will always scan the map in full.

      The source leverages data locality by making each of the underlying processors fetch only those entries that are stored on the member where it is running.

      The source does not save any state to snapshot. If the job is restarted, it will re-emit all entries.

      If the `IMap` is modified while being read, or if there is a cluster topology change (triggering data migration), the source may miss and/or duplicate some entries.

      The default local parallelism for this processor is 2 (or 1 if just 1 CPU is available).

      #### Predicate/projection class requirements

      The classes implementing `predicate` and `projection` need to be available on the cluster's classpath or loaded using Hazelcast User Code Deployment. It's not enough to add them to the job classpath in `JobConfig`. The same is true for the class of the objects stored in the map itself. If you cannot meet these requirements, use `map(String)` and add a subsequent `map` or `filter` stage.
      Type Parameters:
        `T` - type of emitted item
      Parameters:
        `mapName` - the name of the map
        `predicate` - the predicate to filter the events. If you want to specify just the projection, use `GenericPredicates.alwaysTrue()` as a pass-through predicate
        `projection` - the projection to map the events. If the projection returns a `null` for an item, that item will be filtered out. If you want to specify just the predicate, use `Projections.identity()`.

    - #### map

          @Nonnull
          public static <T,K,V> BatchSource<T> map(@Nonnull
                                                            IMap<? extends K,? extends V> map,
                                                            @Nonnull
                                                            Predicate<? super K,? super V> predicate,
                                                            @Nonnull
                                                            Projection<? super Map.Entry<K,V>,? extends T> projection)

      Returns a source that fetches entries from the given Hazelcast `IMap`. By supplying a `predicate` and `projection` here instead of in separate `map/filter` transforms you allow the source to apply these functions early, before generating any output, with the potential of significantly reducing data traffic.

      NOTE: Jet only remembers the name of the map you supply and acquires a map with that name on the local cluster. If you supply a map instance from another cluster, no error will be thrown to indicate this.

      If your data is stored in the IMDG using the portable serialization format, there are additional optimizations available when using `Projections.singleAttribute(java.lang.String)` and `Projections.multiAttribute(java.lang.String...)`) to create your projection instance and using the `GenericPredicates` factory or `PredicateBuilder` to create the predicate. In this case Jet can test the predicate and apply the projection without deserializing the whole object.

      Due to the current limitations in the way Jet reads the map it can't use any indexes on the map. It will always scan the map in full.

      The source leverages data locality by making each of the underlying processors fetch only those entries that are stored on the member where it is running.

      The source does not save any state to snapshot. If the job is restarted, it will re-emit all entries.

      If the `IMap` is modified while being read, or if there is a cluster topology change (triggering data migration), the source may miss and/or duplicate some entries.

      The default local parallelism for this processor is 2 (or 1 if just 1 CPU is available).

      #### Predicate/projection class requirements

      The classes implementing `predicate` and `projection` need to be available on the cluster's classpath or loaded using Hazelcast User Code Deployment. It's not enough to add them to the job classpath in `JobConfig`. The same is true for the class of the objects stored in the map itself. If you cannot meet these requirements, use `map(String)` and add a subsequent `map` or `filter` stage.

      Type Parameters:
        `T` - type of emitted item
      Parameters:
        `map` - the Hazelcast map to draw data from
        `predicate` - the predicate to filter the events. If you want to specify just the projection, use `GenericPredicates.alwaysTrue()` as a pass-through predicate
        `projection` - the projection to map the events. If the projection returns a `null` for an item, that item will be filtered out. If you want to specify just the predicate, use `Projections.identity()`.

    - #### map

          @Nonnull
          public static <T,K,V> BatchSource<T> map(@Nonnull
                                                            String mapName,
                                                            @Nonnull
                                                            Predicate<? super K,? super V> predicate,
                                                            @Nonnull
                                                            DistributedFunction<? super Map.Entry<K,V>,? extends T> projectionFn)

      Convenience for `map(String, Predicate, Projection)` which uses a `DistributedFunction` as the projection function.

    - #### map

          @Nonnull
          public static <T,K,V> BatchSource<T> map(@Nonnull
                                                            IMap<? extends K,? extends V> map,
                                                            @Nonnull
                                                            Predicate<? super K,? super V> predicate,
                                                            @Nonnull
                                                            DistributedFunction<? super Map.Entry<K,V>,? extends T> projectionFn)

      Convenience for `map(IMap, Predicate, Projection)` which uses a `DistributedFunction` as the projection function.

      NOTE: Jet only remembers the name of the map you supply and acquires a map with that name on the local cluster. If you supply a map instance from another cluster, no error will be thrown to indicate this.

    - #### mapJournal

          @Nonnull
          public static <T,K,V> StreamSource<T> mapJournal(@Nonnull
                                                                    String mapName,
                                                                    @Nonnull
                                                                    DistributedPredicate<? super EventJournalMapEvent<K,V>> predicateFn,
                                                                    @Nonnull
                                                                    DistributedFunction<? super EventJournalMapEvent<K,V>,? extends T> projectionFn,
                                                                    @Nonnull
                                                                    JournalInitialPosition initialPos)

      Returns a source that will stream `EventJournalMapEvent`s of the Hazelcast `IMap` with the specified name. By supplying a `predicate` and `projection` here instead of in separate `map/filter` transforms you allow the source to apply these functions early, before generating any output, with the potential of significantly reducing data traffic.

      The source leverages data locality by making each of the underlying processors fetch only those entries that are stored on the member where it is running.

      To use an `IMap` as a streaming source, you must `configure the event journal` for it. The journal has fixed capacity and will drop events if it overflows.

      The source saves the journal offset to the snapshot. If the job restarts, it starts emitting from the saved offset with an exactly-once guarantee (unless the journal has overflowed).

      The default local parallelism for this processor is 2 (or 1 if just 1 CPU is available).

      #### Predicate/projection class requirements

      The classes implementing `predicateFn` and `projectionFn` need to be available on the cluster's classpath or loaded using Hazelcast User Code Deployment. It's not enough to add them to the job classpath in `JobConfig`. The same is true for the class of the objects stored in the map itself. If you cannot meet these requirements, use `mapJournal(String, JournalInitialPosition)` and add a subsequent `map` or `filter` stage.
      Type Parameters:
        `T` - type of emitted item
      Parameters:
        `mapName` - the name of the map
        `predicateFn` - the predicate to filter the events. If you want to specify just the projection, use `Util.mapPutEvents()` to pass only `ADDED` and `UPDATED` events.
        `projectionFn` - the projection to map the events. If the projection returns a `null` for an item, that item will be filtered out. You may use `Util.mapEventToEntry()` to extract just the key and the new value.
        `initialPos` - describes which event to start receiving from

    - #### mapJournal

          @Nonnull
          public static <T,K,V> StreamSource<T> mapJournal(@Nonnull
                                                                    IMap<? extends K,? extends V> map,
                                                                    @Nonnull
                                                                    DistributedPredicate<? super EventJournalMapEvent<K,V>> predicateFn,
                                                                    @Nonnull
                                                                    DistributedFunction<? super EventJournalMapEvent<K,V>,? extends T> projectionFn,
                                                                    @Nonnull
                                                                    JournalInitialPosition initialPos)

      Returns a source that will stream `EventJournalMapEvent`s of the given Hazelcast `IMap`. By supplying a `predicate` and `projection` here instead of in separate `map/filter` transforms you allow the source to apply these functions early, before generating any output, with the potential of significantly reducing data traffic.

      NOTE: Jet only remembers the name of the map you supply and acquires a map with that name on the local cluster. If you supply a map instance from another cluster, no error will be thrown to indicate this.

      The source leverages data locality by making each of the underlying processors fetch only those entries that are stored on the member where it is running.

      To use an `IMap` as a streaming source, you must `configure the event journal` for it. The journal has fixed capacity and will drop events if it overflows.

      The source saves the journal offset to the snapshot. If the job restarts, it starts emitting from the saved offset with an exactly-once guarantee (unless the journal has overflowed).

      The default local parallelism for this processor is 2 (or 1 if just 1 CPU is available).

      #### Predicate/projection class requirements

      The classes implementing `predicateFn` and `projectionFn` need to be available on the cluster's classpath or loaded using Hazelcast User Code Deployment. It's not enough to add them to the job classpath in `JobConfig`. The same is true for the class of the objects stored in the map itself. If you cannot meet these requirements, use `mapJournal(String, JournalInitialPosition)` and add a subsequent `map` or `filter` stage.
      Type Parameters:
        `T` - type of emitted item
      Parameters:
        `map` - the map to draw data from
        `predicateFn` - the predicate to filter the events. If you want to specify just the projection, use `Util.mapPutEvents()` to pass only `ADDED` and `UPDATED` events.
        `projectionFn` - the projection to map the events. If the projection returns a `null` for an item, that item will be filtered out. You may use `Util.mapEventToEntry()` to extract just the key and the new value.
        `initialPos` - describes which event to start receiving from

    - #### mapJournal

          @Nonnull
          public static <K,V> StreamSource<Map.Entry<K,V>> mapJournal(@Nonnull
                                                                               String mapName,
                                                                               @Nonnull
                                                                               JournalInitialPosition initialPos)

      Convenience for `mapJournal(String, DistributedPredicate, DistributedFunction, JournalInitialPosition)` which will pass only `ADDED` and `UPDATED` events and will project the event's key and new value into a `Map.Entry`.

    - #### mapJournal

          @Nonnull
          public static <K,V> StreamSource<Map.Entry<K,V>> mapJournal(@Nonnull
                                                                               IMap<? extends K,? extends V> map,
                                                                               @Nonnull
                                                                               JournalInitialPosition initialPos)

      Convenience for `mapJournal(IMap, DistributedPredicate, DistributedFunction, JournalInitialPosition)` which will pass only `ADDED` and `UPDATED` events and will project the event's key and new value into a `Map.Entry`.

      NOTE: Jet only remembers the name of the map you supply and acquires a map with that name on the local cluster. If you supply a map instance from another cluster, no error will be thrown to indicate this.

    - #### remoteMap

          @Nonnull
          public static <K,V> BatchSource<Map.Entry<K,V>> remoteMap(@Nonnull
                                                                             String mapName,
                                                                             @Nonnull
                                                                             ClientConfig clientConfig)

      Returns a source that fetches entries from the Hazelcast `IMap` with the specified name in a remote cluster identified by the supplied `ClientConfig` and emits them as `Map.Entry`.

      The source does not save any state to snapshot. If the job is restarted, it will re-emit all entries.

      If the `IMap` is modified while being read, or if there is a cluster topology change (triggering data migration), the source may miss and/or duplicate some entries.

      The default local parallelism for this processor is 1.

    - #### remoteMap

          @Nonnull
          public static <T,K,V> BatchSource<T> remoteMap(@Nonnull
                                                                  String mapName,
                                                                  @Nonnull
                                                                  ClientConfig clientConfig,
                                                                  @Nonnull
                                                                  Predicate<? super K,? super V> predicate,
                                                                  @Nonnull
                                                                  Projection<? super Map.Entry<K,V>,? extends T> projection)

      Returns a source that fetches entries from a remote Hazelcast `IMap` with the specified name in a remote cluster identified by the supplied `ClientConfig`. By supplying a `predicate` and `projection` here instead of in separate `map/filter` transforms you allow the source to apply these functions early, before generating any output, with the potential of significantly reducing data traffic. If your data is stored in the IMDG using the portable serialization format, there are additional optimizations available when using `Projections.singleAttribute()` and `Projections.multiAttribute()`) to create your projection instance and using the `GenericPredicates` factory or `PredicateBuilder` to create the predicate. In this case Jet can test the predicate and apply the projection without deserializing the whole object.

      Due to the current limitations in the way Jet reads the map it can't use any indexes on the map. It will always scan the map in full.

      The source does not save any state to snapshot. If the job is restarted, it will re-emit all entries.

      If the `IMap` is modified while being read, or if there is a cluster topology change (triggering data migration), the source may miss and/or duplicate some entries.

      The default local parallelism for this processor is 1.

      #### Predicate/projection class requirements

      The classes implementing `predicate` and `projection` need to be available on the remote cluster's classpath or loaded using Hazelcast User Code Deployment. It's not enough to add them to the job classpath in `JobConfig`. The same is true for the class of the objects stored in the map itself. If you cannot meet these conditions, use `remoteMap(String, ClientConfig)` and add a subsequent `map` or `filter` stage.
      Type Parameters:
        `T` - type of emitted item
      Parameters:
        `mapName` - the name of the map
        `predicate` - the predicate to filter the events. If you want to specify just the projection, use `GenericPredicates.alwaysTrue()` as a pass-through predicate
        `projection` - the projection to map the events. If the projection returns a `null` for an item, that item will be filtered out. If you want to specify just the predicate, use `Projections.identity()`.

    - #### remoteMap

          @Nonnull
          public static <T,K,V> BatchSource<T> remoteMap(@Nonnull
                                                                  String mapName,
                                                                  @Nonnull
                                                                  ClientConfig clientConfig,
                                                                  @Nonnull
                                                                  Predicate<? super K,? super V> predicate,
                                                                  @Nonnull
                                                                  DistributedFunction<? super Map.Entry<K,V>,? extends T> projectionFn)

      Convenience for `remoteMap(String, ClientConfig, Predicate, Projection)` which use a `DistributedFunction` as the projection function.

    - #### remoteMapJournal

          @Nonnull
          public static <T,K,V> StreamSource<T> remoteMapJournal(@Nonnull
                                                                          String mapName,
                                                                          @Nonnull
                                                                          ClientConfig clientConfig,
                                                                          @Nonnull
                                                                          DistributedPredicate<? super EventJournalMapEvent<K,V>> predicateFn,
                                                                          @Nonnull
                                                                          DistributedFunction<? super EventJournalMapEvent<K,V>,? extends T> projectionFn,
                                                                          @Nonnull
                                                                          JournalInitialPosition initialPos)

      Returns a source that will stream the `EventJournalMapEvent` events of the Hazelcast `IMap` with the specified name from a remote cluster. By supplying a `predicate` and `projection` here instead of in separate `map/filter` transforms you allow the source to apply these functions early, before generating any output, with the potential of significantly reducing data traffic.

      To use an `IMap` as a streaming source, you must `configure the event journal` for it. The journal has fixed capacity and will drop events if it overflows.

      The source saves the journal offset to the snapshot. If the job restarts, it starts emitting from the saved offset with an exactly-once guarantee (unless the journal has overflowed).

      The default local parallelism for this processor is 1.

      #### Predicate/projection class requirements

      The classes implementing `predicateFn` and `projectionFn` need to be available on the remote cluster's classpath or loaded using Hazelcast User Code Deployment. It's not enough to add them to the job classpath in `JobConfig`. The same is true for the class of the objects stored in the map itself. If you cannot meet these requirements, use `remoteMapJournal(String, ClientConfig, JournalInitialPosition)` and add a subsequent `map` or `filter` stage.
      Type Parameters:
        `K` - type of key
        `V` - type of value
        `T` - type of emitted item
      Parameters:
        `mapName` - the name of the map
        `clientConfig` - configuration for the client to connect to the remote cluster
        `predicateFn` - the predicate to filter the events. You may use `Util.mapPutEvents()` to pass only `ADDED` and `UPDATED` events.
        `projectionFn` - the projection to map the events. If the projection returns a `null` for an item, that item will be filtered out. You may use `Util.mapEventToEntry()` to extract just the key and the new value.
        `initialPos` - describes which event to start receiving from

    - #### remoteMapJournal

          @Nonnull
          public static <K,V> StreamSource<Map.Entry<K,V>> remoteMapJournal(@Nonnull
                                                                                     String mapName,
                                                                                     @Nonnull
                                                                                     ClientConfig clientConfig,
                                                                                     @Nonnull
                                                                                     JournalInitialPosition initialPos)

      Convenience for `remoteMapJournal(String, ClientConfig, DistributedPredicate, DistributedFunction, JournalInitialPosition)` which will pass only `ADDED` and `UPDATED` events and will project the event's key and new value into a `Map.Entry`.

    - #### cache

          @Nonnull
          public static <K,V> BatchSource<Map.Entry<K,V>> cache(@Nonnull
                                                                         String cacheName)

      Returns a source that fetches entries from a Hazelcast `ICache` with the given name and emits them as `Map.Entry`. It leverages data locality by making each of the underlying processors fetch only those entries that are stored on the member where it is running.

      The source does not save any state to snapshot. If the job is restarted, it will re-emit all entries.

      If the `ICache` is modified while being read, or if there is a cluster topology change (triggering data migration), the source may miss and/or duplicate some entries.

      The default local parallelism for this processor is 2 (or 1 if just 1 CPU is available).

    - #### cacheJournal

          @Nonnull
          public static <T,K,V> StreamSource<T> cacheJournal(@Nonnull
                                                                      String cacheName,
                                                                      @Nonnull
                                                                      DistributedPredicate<? super EventJournalCacheEvent<K,V>> predicateFn,
                                                                      @Nonnull
                                                                      DistributedFunction<? super EventJournalCacheEvent<K,V>,? extends T> projectionFn,
                                                                      @Nonnull
                                                                      JournalInitialPosition initialPos)

      Returns a source that will stream the `EventJournalCacheEvent` events of a Hazelcast `ICache` with the specified name. By supplying a `predicate` and `projection` here instead of in separate `map/filter` transforms you allow the source to apply these functions early, before generating any output, with the potential of significantly reducing data traffic.

      The source leverages data locality by making each of the underlying processors fetch only those entries that are stored on the member where it is running.

      To use an `ICache` as a streaming source, you must `configure the event journal` for it. The journal has fixed capacity and will drop events if it overflows.

      The source saves the journal offset to the snapshot. If the job restarts, it starts emitting from the saved offset with an exactly-once guarantee (unless the journal has overflowed).

      The default local parallelism for this processor is 2 (or 1 if just 1 CPU is available).

      #### Predicate/projection class requirements

      The classes implementing `predicateFn` and `projectionFn` need to be available on the cluster's classpath or loaded using Hazelcast User Code Deployment. It's not enough to add them to the job classpath in `JobConfig`. The same is true for the class of the objects stored in the cache itself. If you cannot meet these conditions, use `cacheJournal(String, JournalInitialPosition)` and add a subsequent `map` or `filter` stage.
      Type Parameters:
        `T` - type of emitted item
      Parameters:
        `cacheName` - the name of the cache
        `predicateFn` - the predicate to filter the events. You may use `Util.cachePutEvents()` to pass only `CREATED` and `UPDATED` events.
        `projectionFn` - the projection to map the events. If the projection returns a `null` for an item, that item will be filtered out. You may use `Util.cacheEventToEntry()` to extract just the key and the new value.
        `initialPos` - describes which event to start receiving from

    - #### cacheJournal

          @Nonnull
          public static <K,V> StreamSource<Map.Entry<K,V>> cacheJournal(@Nonnull
                                                                                 String cacheName,
                                                                                 @Nonnull
                                                                                 JournalInitialPosition initialPos)

      Convenience for `cacheJournal(String, DistributedPredicate, DistributedFunction, JournalInitialPosition)` which will pass only `CREATED` and `UPDATED` events and will project the event's key and new value into a `Map.Entry`.

    - #### remoteCache

          @Nonnull
          public static <K,V> BatchSource<Map.Entry<K,V>> remoteCache(@Nonnull
                                                                               String cacheName,
                                                                               @Nonnull
                                                                               ClientConfig clientConfig)

      Returns a source that fetches entries from the Hazelcast `ICache` with the specified name in a remote cluster identified by the supplied `ClientConfig` and emits them as `Map.Entry`.

      The source does not save any state to snapshot. If the job is restarted, it will re-emit all entries.

      If the `ICache` is modified while being read, or if there is a cluster topology change (triggering data migration), the source may miss and/or duplicate some entries.

      The default local parallelism for this processor is 1.

    - #### remoteCacheJournal

          @Nonnull
          public static <T,K,V> StreamSource<T> remoteCacheJournal(@Nonnull
                                                                            String cacheName,
                                                                            @Nonnull
                                                                            ClientConfig clientConfig,
                                                                            @Nonnull
                                                                            DistributedPredicate<? super EventJournalCacheEvent<K,V>> predicateFn,
                                                                            @Nonnull
                                                                            DistributedFunction<? super EventJournalCacheEvent<K,V>,? extends T> projectionFn,
                                                                            @Nonnull
                                                                            JournalInitialPosition initialPos)

      Returns a source that will stream the `EventJournalCacheEvent` events of the Hazelcast `ICache` with the specified name from a remote cluster. By supplying a `predicate` and `projection` here instead of in separate `map/filter` transforms you allow the source to apply these functions early, before generating any output, with the potential of significantly reducing data traffic.

      To use an `ICache` as a streaming source, you must `configure the event journal` for it. The journal has fixed capacity and will drop events if it overflows.

      The source saves the journal offset to the snapshot. If the job restarts, it starts emitting from the saved offset with an exactly-once guarantee (unless the journal has overflowed).

      The default local parallelism for this processor is 1.

      #### Predicate/projection class requirements

      The classes implementing `predicateFn` and `projectionFn` need to be available on the cluster's classpath or loaded using Hazelcast User Code Deployment. It's not enough to add them to the job classpath in `JobConfig`. The same is true for the class of the objects stored in the cache itself. If you cannot meet these conditions, use `remoteCacheJournal(String, ClientConfig, JournalInitialPosition)` and add a subsequent `map` or `filter` stage.
      Type Parameters:
        `T` - type of emitted item
      Parameters:
        `cacheName` - the name of the cache
        `clientConfig` - configuration for the client to connect to the remote cluster
        `predicateFn` - the predicate to filter the events. You may use `Util.cachePutEvents()` to pass only `CREATED` and `UPDATED` events.
        `projectionFn` - the projection to map the events. If the projection returns a `null` for an item, that item will be filtered out. You may use `Util.cacheEventToEntry()` to extract just the key and the new value.
        `initialPos` - describes which event to start receiving from

    - #### remoteCacheJournal

          @Nonnull
          public static <K,V> StreamSource<Map.Entry<K,V>> remoteCacheJournal(@Nonnull
                                                                                       String cacheName,
                                                                                       @Nonnull
                                                                                       ClientConfig clientConfig,
                                                                                       @Nonnull
                                                                                       JournalInitialPosition initialPos)

      Convenience for `remoteCacheJournal(String, ClientConfig, DistributedPredicate, DistributedFunction, JournalInitialPosition)` which will pass only `CREATED` and `UPDATED` events and will project the event's key and new value into a `Map.Entry`.

    - #### list

          @Nonnull
          public static <T> BatchSource<T> list(@Nonnull
                                                         String listName)

      Returns a source that emits items retrieved from a Hazelcast `IList`. All elements are emitted on a single member — the one where the entire list is stored by the IMDG.

      If the `IList` is modified while being read, the source may miss and/or duplicate some entries.

      The source does not save any state to snapshot. If the job is restarted, it will re-emit all entries.

      The default local parallelism for this processor is 1.

    - #### list

          @Nonnull
          public static <T> BatchSource<T> list(@Nonnull
                                                         IList<? extends T> list)

      Returns a source that emits items retrieved from a Hazelcast `IList`. All elements are emitted on a single member — the one where the entire list is stored by the IMDG.

      If the `IList` is modified while being read, the source may miss and/or duplicate some entries.

      NOTE: Jet only remembers the name of the list you supply and acquires a list with that name on the local cluster. If you supply a list instance from another cluster, no error will be thrown to indicate this.

      The source does not save any state to snapshot. If the job is restarted, it will re-emit all entries.

      The default local parallelism for this processor is 1.

    - #### remoteList

          @Nonnull
          public static <T> BatchSource<T> remoteList(@Nonnull
                                                               String listName,
                                                               @Nonnull
                                                               ClientConfig clientConfig)

      Returns a source that emits items retrieved from a Hazelcast `IList` in a remote cluster identified by the supplied `ClientConfig`. All elements are emitted on a single member.

      If the `IList` is modified while being read, the source may miss and/or duplicate some entries.

      The source does not save any state to snapshot. If the job is restarted, it will re-emit all entries.

      The default local parallelism for this processor is 1.

    - #### socket

          @Nonnull
          public static StreamSource<String> socket(@Nonnull
                                                             String host,
                                                             int port,
                                                             @Nonnull
                                                             Charset charset)

      Returns a source which connects to the specified socket and emits lines of text received from it. It decodes the text using the supplied `charset`.

      Each underlying processor opens its own TCP connection, so there will be `clusterSize * localParallelism` open connections to the server.

      The source completes when the server closes the socket. It never attempts to reconnect. Any `IOException` will cause the job to fail.

      The source does not save any state to snapshot. On job restart, it will emit whichever items the server sends. The implementation uses non-blocking API, the processor is cooperative.

      The default local parallelism for this processor is 1.

    - #### socket

          @Nonnull
          public static StreamSource<String> socket(@Nonnull
                                                             String host,
                                                             int port)

      Convenience for `socket(host, port, charset)` with UTF-8 as the charset.
      Parameters:
        `host` - the hostname to connect to
        `port` - the port to connect to

    - #### filesBuilder

          @Nonnull
          public static FileSourceBuilder filesBuilder(@Nonnull
                                                                String directory)

      Returns a builder object that offers a step-by-step fluent API to build a custom source to read files for the Pipeline API. The source reads lines from files in a directory (but not its subdirectories). Using this builder you can build batching or streaming reader.

    - #### files

          @Nonnull
          public static BatchSource<String> files(@Nonnull
                                                           String directory)

      A source to read all files in a directory in a batch way.

      This method is a shortcut for:

             filesBuilder(directory)
                .charset(UTF_8)
                .glob(GLOB_WILDCARD)
                .sharedFileSystem(false)
                .mapToOutputFn((fileName, line) -> line)
                .build()

      If files are appended to while being read, the addition might or might not be emitted or part of a line can be emitted. If files are modified in more complex ways, the behavior is undefined. See `filesBuilder(String)`.

    - #### fileWatcher

          @Nonnull
          public static StreamSource<String> fileWatcher(@Nonnull
                                                                  String watchedDirectory)

      A source to stream lines added to files in a directory. This is a streaming source, it will watch directory and emit lines as they are appended to files in that directory.

      This method is a shortcut for:

             filesBuilder(directory)
                .charset(UTF_8)
                .glob(GLOB_WILDCARD)
                .sharedFileSystem(false)
                .mapToOutputFn((fileName, line) -> line)
                .buildWatcher()

      See `filesBuilder(String)`.

    - #### jmsQueue

          @Nonnull
          public static StreamSource<javax.jms.Message> jmsQueue(@Nonnull
                                                                          DistributedSupplier<? extends javax.jms.ConnectionFactory> factorySupplier,
                                                                          @Nonnull
                                                                          String name)

      Convenience for `jmsQueueBuilder(DistributedSupplier)`. This version creates a connection without any authentication parameters and uses non-transacted sessions with `Session.AUTO_ACKNOWLEDGE` mode. JMS `Message` objects are emitted to downstream.
      Parameters:
        `factorySupplier` - supplier to obtain JMS connection factory
        `name` - the name of the queue

    - #### jmsQueueBuilder

          @Nonnull
          public static JmsSourceBuilder jmsQueueBuilder(DistributedSupplier<? extends javax.jms.ConnectionFactory> factorySupplier)

      Returns a builder object that offers a step-by-step fluent API to build a custom JMS `StreamSource` for the Pipeline API. See javadoc on `JmsSourceBuilder` methods for more details.

      The source does not save any state to snapshot. The source starts emitting items where it left from.

      IO failures should be handled by the JMS provider. If any JMS operation throws an exception, the job will fail. Most of the providers offer a configuration parameter to enable auto-reconnection, refer to provider documentation for details.

      Default local parallelism for this processor is 4 (or less if less CPUs are available).

    - #### jmsTopic

          @Nonnull
          public static StreamSource<javax.jms.Message> jmsTopic(@Nonnull
                                                                          DistributedSupplier<? extends javax.jms.ConnectionFactory> factorySupplier,
                                                                          @Nonnull
                                                                          String name)

      Convenience for `jmsTopicBuilder(DistributedSupplier)`. This version creates a connection without any authentication parameters and uses non-transacted sessions with `Session.AUTO_ACKNOWLEDGE` mode. JMS `Message` objects are emitted to downstream.
      Parameters:
        `factorySupplier` - supplier to obtain JMS connection factory
        `name` - the name of the topic

    - #### jmsTopicBuilder

          @Nonnull
          public static JmsSourceBuilder jmsTopicBuilder(DistributedSupplier<? extends javax.jms.ConnectionFactory> factorySupplier)

      Returns a builder object that offers a step-by-step fluent API to build a custom JMS `StreamSource` for the Pipeline API. See javadoc on `JmsSourceBuilder` methods for more details.

      Topic is a non-distributed source: if messages are consumed by multiple consumers, all of them will get the same messages. Therefore the source operates on a single member and with local parallelism of 1. Setting local parallelism to a value other than 1 causes an `IllegalArgumentException`.

      The source does not save any state to snapshot. Behavior of job restart changes according to the consumer. If it is a durable consumer and a unique client identifier is set for the connection then JMS provider persists items during restart and the source starts where it left from. If the consumer is non-durable then source emits the items published after the restart.

      IO failures should be handled by the JMS provider. If any JMS operation throws an exception, the job will fail. Most of the providers offer a configuration parameter to enable auto-reconnection, refer to provider documentation for details.

    - #### jdbc

          public static <T> BatchSource<T> jdbc(@Nonnull
                                                DistributedSupplier<? extends Connection> connectionSupplier,
                                                @Nonnull
                                                ToResultSetFunction resultSetFn,
                                                @Nonnull
                                                DistributedFunction<? super ResultSet,? extends T> createOutputFn)

      Returns a source which connects to the specified database using the given `connectionSupplier`, queries the database and creates a result set using the the given `resultSetFn`. It creates output objects from the `ResultSet` using given `mapOutputFn` and emits them to downstream.

      `resultSetFn` gets the created connection, total parallelism (local parallelism * member count) and global processor index as arguments and produces a result set. The parallelism and processor index arguments should be used to fetch a part of the whole result set specific to the processor. If the table itself isn't partitioned by the same key, then running multiple queries might not really be faster than using the simpler version of this method, do your own testing.

      `createOutputFn` gets the `ResultSet` and creates desired output object. The function is called for each row of the result set, user should not call `ResultSet.next()` or any other cursor-navigating functions.

      Example:

               p.drawFrom(Sources.jdbc(
                   () -> DriverManager.getConnection(DB_CONNECTION_URL),
                   (con, parallelism, index) -> {
                        PreparedStatement stmt = con.prepareStatement("SELECT * FROM TABLE WHERE MOD(id, ?) = ?)");
                        stmt.setInt(1, parallelism);
                        stmt.setInt(2, index);
                        return stmt.executeQuery();
                   },
                   resultSet -> new Person(resultSet.getInt(1), resultSet.getString(2))))

      If the underlying table is modified while being read, the source may miss and/or duplicate some entries, because multiple queries for parts of the data on multiple members will be executed.

      The source does not save any state to snapshot. If the job is restarted, it will re-emit all entries.

      Any `SQLException` will cause the job to fail.

      The default local parallelism for this processor is 1.

      Type Parameters:
        `T` - type of output objects
      Parameters:
        `connectionSupplier` - creates the connection
        `resultSetFn` - creates a `ResultSet` using the connection, total parallelism and index
        `createOutputFn` - creates output objects from `ResultSet`

    - #### jdbc

          public static <T> BatchSource<T> jdbc(@Nonnull
                                                String connectionURL,
                                                @Nonnull
                                                String query,
                                                @Nonnull
                                                DistributedFunction<? super ResultSet,? extends T> createOutputFn)

      Convenience for `jdbc(DistributedSupplier, ToResultSetFunction, DistributedFunction)`. A non-distributed, single-worker source which fetches the whole resultSet with a single query on single member.

      This method executes exactly one query in the target database. If the underlying table is modified while being read, the behavior depends on the configured transaction isolation level in the target database. Refer to the documentation for the target database system.

      Example:

               p.drawFrom(Sources.jdbc(
                   DB_CONNECTION_URL,
                   "select ID, NAME from PERSON",
                   resultSet -> new Person(resultSet.getInt(1), resultSet.getString(2))))

Copyright © 2018. All rights reserved.