Hazelcast Jet JobConfig Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.jet.config.JobConfig;

JobConfig nesnesi JetService nesnesine parametre olarak geçilir.

addClass metodu

Neden Lazım?

Çünkü lambda'lar otomatik gönderilmiyor. Şu kod çalışmaz. Lambda nesnesi HazelcastSerializationException'a sebep olur.

Pipeline pipeline = Pipeline.create()
  .readFrom(Sources.jdbc(
    url,
    "SELECT * FROM person", 
    r-> new Person(rs.getInt(1),rs.getString(2),rs.getString(3))
    )
  )
  .writeTo(Sinks.logger())
  .getPipeline();

JobConfig jobConfig = new JobConfig();
  .addClass(Person.class);
  setName("example")

hazelcastInstance.getJet().newJob(pipeline,jobConfig);

Örnek

Şu kod çalışır.

BatchSource<String> items = TestSources.items(
  "United States", "Turkey", "United Kingdom", "Poland", "Ukraine"
);

Pipeline pipeline = Pipeline.create()
  .readFrom(items)
  .map(new UpperCaseFunction())
  .writeTo(Sinks.logger())
  .getPipeline();

JobConfig jobConfig = new JobConfig()
  .addClass(UpperCaseFunction.class);

client.getJet().newJob(pipeline, jobConfig);

Örnek

Şöyle yaparız

jobConfig
  .setName("Myjob")
  .addPackage("com.foo")
  .addClass(com.hazelcast.jet.cdc.DebeziumCdcSources.class)
  .addClass(io.debezium.connector.mongodb.MongoDbConnector.class);
...
hz.getJet().newJobIfAbsent(pipeline, jobConfig);

addCustomClasspath metodu

Member başlarken ClusterProperty.PROCESSOR_CUSTOM_LIB_DIR ile belirtilen dizindeki kendi oluşturduğum bir jar dosyasını Job'ın kullanabilmesini sağlar. 

addJar metodu

Açıklaması şöyle

Adds a JAR whose contents will be accessible to all the code attached to the

underlying pipeline or DAG, but not to any other code. An important example

is the IMap data source, which can instantiate only the classes from

the Jet instance's classpath.)

This variant identifies the JAR with a path string. The filename part will be

used as the resource ID, so two JARs with the same filename will be in

conflict.

Cannot be used for JetService#newLightJob(Pipeline) light jobs

@implNote Backing storage for this method is an {@link IMap} with a default

backup count of 1. When adding big files as a resource, size the

cluster accordingly in terms of memory, since each file will have 2

copies inside the cluster(primary + backup replica).

Örnek

Şöyle yaparız

String jarPath = "/mydir/foo.jar";
jobConfig.addJar(jarPath);

addInZip metodu

Örnek

Şöyle yaparız

URL url = new URL("https://.../foo.zip");
jobConfig.addJarsInZip(url);

registerSerializer metodu

İmzası şöyle. Yani ikinci parametre bir StreamSerializer

public <T, S extends StreamSerializer<T>> JobConfig registerSerializer(
  Class<T> clazz,
  Class<S> serializerClass
)

Örnek

Şöyle yaparız

new JobConfig()
    .registerSerializer(Person.class, PersonSerializer.class)

setArgument metodu

İş çalıştırılırken kullanılacak parametreyi belirtir

setName metodu

Şöyle yaparız

jobConfig
  .setName("Myjob")
  ...

setProcessingGuarantee metodu

Şöyle yaparız

JobConfig jobConfig = new JobConfig()
  .setName(...)
  .setProcessingGuarantee(ProcessingGuarantee.AT_LEAST_ONCE)
  .setSnapshotIntervalMillis(250)
  .setSuspendOnFailure(true);

setTimeoutMillis metodu

İş belirtilen sürede bitmezse exception fırlatır. Şöyle yaparız

JobConfig jobConfig = new JobConfig()
  .setTimeoutMillis(10L);

HazelcastAPI MapStore Arayüzü - Harici Bir Kaynağa Veri Yazar

Giriş

Şu satırı dahil ederiz

import com.hazelcast.map.MapStore;

MapStore vs Tiered Storage

Eğer Hazelcast verisini veri tabanı yerine Hazelcast'in kendisinin saklamasını istiyorsak Tiered Storage özelliği kullanılabilir. Açıklaması şöyle

Hazelcast Platform 5.1 introduces a new feature, “Tiered Storage.”

You can read about it here.

Bu açıklamaya göre veri tabanından kurtulmak mümkün olabilir. 

What doesn’t seem right is the technical debt. You need Hazelcast and something else, use a map store to connect them, and perhaps have licenses and staffing costs to run the “something else.” In many cases, the something else is a relational database when all you really want is disk storage.

MapStore Kullanımı

Normalde IMap arayüzü uçucudur (volatile), ancak IMap'i  veri tabanına kaydetmek istersek MapStore arayüzünü takmak gerekir. MapStore arayüzü MapLoader'dan kalıtır. Bu yüzden veri tabanından okuma ve yazma işlemlerini yapar. Şeklen şöyle

XML ayarlarının açıklaması şöyle

class-name: Name of the class implementing MapLoader and/or MapStore.

write-delay-seconds: Number of seconds to delay to call the MapStore.store(key, value)`. If the value is zero then it is write-through, so the MapStore.store(key,value) method is called as soon as the entry is updated. Otherwise, it is write-behind; so the updates will be stored after the write-delay-seconds value by calling the Hazelcast.storeAll(map) method. Its default value is 0 (write-through).

write-batch-size: Used to create batch chunks when writing to the external data store. In default mode, all map entries are tried to be written in one go. To create batch chunks, the minimum meaningful value for write-batch-size is 2. For values smaller than 2, it works as in default mode.

write-coalescing: In write-behind mode, Hazelcast coalesces updates on a specific key by default; it applies only the last update on it. You can set this element to false to store all updates performed on a key to the data store.

enabled: True to enable this map-store, false to disable. Its default value is true.

initial-mode: Sets the initial load mode. LAZY is the default load mode, where load is asynchronous. EAGER means map operations are blocked until all partitions are loaded.

1. MapStore sınıfını kullanacaksak veri tabanı bağlantı ve SQL işlerini kendimiz hallederiz.

2. Hazelcast açılırken varsayılan initial-mode değeri LAZY olduğu için arka planda şu tetiklenir

public Iterable<Foo> loadAllKeys()

3. IMap.put() metodunu çağırırsak şu tetiklenir.

public void store(Foo key, Bar value) 

4. IMap.get() metodunu çağırırsak ve key bellekte yok şu tetiklenir.

public void delete(Foo key) {

Kendi Sınıfım

Örnek

Şöyle yaparız

<map name="myMap">
  <backup-count>1</backup-count>
  <time-to-live-seconds>0</time-to-live-seconds>
  <max-idle-seconds>0</max-idle-seconds>

  <map-store enabled="true">
    <class-name>foo.bar.MyMapLoader</class-name>
    <write-delay-seconds>0</write-delay-seconds>
  </map-store>
</map>

Örnek

Şöyle yaparız

<hazelcast>
    ...
    <map name="supplements">
        <map-store enabled="true" initial-mode="LAZY">
            <class-name>com.hazelcast.loader.YourMapStoreImplementation</class-name>
            <properties>
                <property name="mongo.url">mongodb://localhost:27017</property>
                <property name="mongo.db">mydb</property>
                <property name="mongo.collection">supplements</property>
            </properties>
        </map-store>
    </map>
    ...
</hazelcast>

public class YourMapStoreImplementation implements 
  MapStore<String, Supplement>, MapLoaderLifecycleSupport {
  ...
}

GenericMapStore Sınıfı

Eğer value alanında GenericRecord olan bir map varsa bunu veri tabanına kaydetmek için Hazelcast ile hazır gelen GenericMapStore kullanılabilir. 

IMap<Integer, GenericRecord> map = hazelcastInstance.getMap("..");

Jet Fault Tolerance

Giriş

Açıklaması şöyle

How Jet Fault Tolerance Works

• Regular backups (snapshots)

• Restart computation from last snapshot if topology changes

• Single node failures/additions - snapshot restored from replicated memory storage

Preconditions:

• Replayable source (e.g Kafka, Hazelcast Event Journal, JMS)

• Deterministic (no mutable lookup tables, no randomness)

• Idempotent sink (Upsert or distributed transactions: JMS, JDBC, Kafka)

Snapshot

Açıklaması şöyle. Yani Job state IMap üzerinde saklanır. Job çökerse en son yedek (snapshot) alınan yerden devam edebilir. Bu yöntem aynı zamanda Job'ların ölçeklendirilmesi için de kullanılır

In Hazelcast Jet 0.5, we introduced fault tolerance for streaming computations. Hazelcast Jet periodically takes snapshots of the state of a running job and stores these snapshots in Hazelcast IMaps. In case of a failure, the job is restarted from the last successful snapshot. 

jet komutu

submit seçeneği

Örnek

Şöyle yaparız

${JET}/bin/jet submit ${LABS}/job-deployment/target/jobdeployment-4.1.jar

Hazelcast Jet CDC MySqlCdcSources

Giriş

Şu satırı dahil ederiz

import com.hazelcast.jet.cdc.mysql.MySqlCdcSources;

Örnek

Şöyle yaparız

StreamSource<ChangeRecord> source = MySqlCdcSources.mysql("source")
  .setDatabaseAddress("127.0.01")
  .setDatabasePort(3306)
  .setDatabaseUser("dbUser")
  .setDatabasePassword("pw")
  .setClusterName("dbServer1")
  .setDatabaseWhitelist("inventory")
  .setTableWhiteList("inventory.customers")
  .build()

Pipeline pipeline = Pipeline.create();
pipeline.readFrom(source)
  .withoutTimestamps()
  .peek()
  .writeTo(CdcSinks.map("customers",
             r-> r.key().toMap().get("id"),
             r-> r.value().toObject(Customer.class).toString()));
JobConfig cfg = new JobConfig();
Jet.bootstrappedInstance().newJob(pipeline,cfg);

Hazelcast Jet TestSources Sınıfı

Giriş
Şu satırı dahil ederiz 

import com.hazelcast.jet.pipeline.test.TestSources;

Açıklaması şöyle

Data generators for quick start and testing

items metodu

Belirtilen listeyi döner

Örnek

Şöyle yaparız

Pipeline pipeline = Pipeline.create();

BatchSource<String> leftSource = TestSources.items("the", "quick", "brown", "fox");
BatchSource<String> rightSource = TestSources.items("jumps", "over", "the", "lazy", "dog");

BatchStage<String> left = pipeline.readFrom(leftSource);
BatchStage<String> right = pipeline.readFrom(rightSource);

left.merge(right)
    .writeTo(Sinks.logger());

Örnek

Şöyle yaparız

List<String> source = new ArrayList<>();
...
Pipeline pipeline = Pipeline.create();
pipeline.readFrom(TestSources.items(source))
  ...

itemStream metodu - itemsPerSecond

Saniyede belirtilen sayı kadar nesne döner.

Örnek

Şöyle yaparız

public static void main(String[] args) {
  Pipeline pipeline = Pipeline.create();
  pipeline.readFrom(TestSources.itemStream(10))
    .withoutTimestamps()
    .filter(event -> event.sequence() % 2 == 0)
    .setName("filter out odd numbers")
    .writeTo(Sinks.logger());

  HazelcastInstance hz = Hazelcast.bootstrappedInstance();
  hz.getJet().newJob(pipeline);
}

itemStream metodu - itemsPerSecond + GeneratorFunction

Örnek

Şöyle yaparız. Aslında bu kötü bir örnek. Burada saniyede 10 tane Person nesnesi yaratılıyor ama hepsi aynı sabit String ile ilklendiriliyor

String item = ...; 
Pipeline pipeline = Pipeline.create();
pipeline.readFrom(TestSources.itemStream(10, (t, u) -> item))
  .withoutTimestamps()
  .rebalance()
  .map(t -> new Person(t))
  .rebalance()
  ...

Hazelcast Jet AggregateOperations Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.jet.aggregate.AggregateOperations;

Şeklen şöyle

counting metodu

Örnek

Şöyle yaparız

Pipeline p = Pipeline.create();
p.readFrom(sourceOfLines())
  .flatMap(line -> traverseArray(line.split("\\W+")))
  .filter(word -> !word.isEmpty())
  .groupingKey(wholeItem())
  .aggregate(counting())
  .writeTo(Sinks.map("counts"));

toList metodu

BatchStage<List<X>> şeklinde bir şey döner

Örnek

Şöyle yaparız

jdbcSource
  .aggregate(AggregateOperations.toList())
  .writeTo(Sinks.json("directory-where-a-single-file-will-be-created"))