HazelcastAPI DataSerializable Arayüzü

Giriş

Şu satırı dahil ederiz

import com.hazelcast.nio.serialization.DataSerializable;

 java.io.Externalizable gibidir. Sadece biraz daha verimli olduğu söyleniyor. com.hazelcast.nio.ObjectDataOutput

com.hazelcast.nio.ObjectDataInput

stream'lerini kullanır

DataSerializable arayüzünden kalıtan sınıfları okuyup yazan sınıf DataSerializableSerializer. 

constructor

Açıklaması şöyle

Classes conforming to DataSerializable should provide a no-arguments constructor.

Örnek

Şöyle yaparız. Burada sınıfı no-argument constructor sağlamamış. Yani biraz yanlış var

class Car implements DataSerializable {

  private String name;
  private String number;

  @Override
  public void writeData(ObjectDataOutput out) throws IOException {
    out.writeUTF(name);
    out.writeUTF(number);
  }

  @Override
  public void readData(ObjectDataInput in) throws IOException {
    name = in.readUTF();
    number = in.readUTF();
  }
}

Örnek - Test

Şöyle yaparız

import com.hazelcast.internal.serialization.InternalSerializationService;
import com.hazelcast.internal.serialization.impl.DefaultSerializationServiceBuilder;
import com.hazelcast.internal.serialization.impl.ObjectDataInputStream;
import com.hazelcast.internal.serialization.impl.ObjectDataOutputStream;

@Test
public void testWriteData() throws Exception {
  Foo foo = ...

  DefaultSerializationServiceBuilder defaultSerializationServiceBuilder = 
    new DefaultSerializationServiceBuilder();
  InternalSerializationService serializationService = defaultSerializationServiceBuilder
    .setUseNativeByteOrder(false)
    .setAllowUnsafe(false)
    .setByteOrder(ByteOrder.BIG_ENDIAN)
    .build();

  ByteArrayOutputStream bout = new ByteArrayOutputStream();
  ObjectDataOutput out = new ObjectDataOutputStream(bout, serializationService);

  foo.writeData(out);

  Foo foo2 = new InsertProcessorSupplier();
  byte[] byteArray = bout.toByteArray();
  ByteArrayInputStream bin = new ByteArrayInputStream(byteArray);
  ObjectDataInput in = new ObjectDataInputStream(bin, serializationService);
  foo2.readData(in);
}

HazelcastClientAPI NearCacheConfig Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.config.NearCacheConfig;

setEvictionConfig metodu

maxSize ile belirtilen değere ulaşınca kullanılacak eviction yöntemini belirtir

Örnek

Şöyle yaparız

EvictionConfig evictionConfig = new EvictionConfig()
  .setMaxSizePolicy(MaxSizePolicy.ENTRY_COUNT)
  .setEvictionPolicy(EvictionPolicy.LRU)
  .setSize(1);

NearCachePreloaderConfig preloaderConfig = new NearCachePreloaderConfig()
  .setEnabled(true)
  .setDirectory("nearcache-example")
  .setStoreInitialDelaySeconds(1)
  .setStoreIntervalSeconds(2);

NearCacheConfig nearCacheConfig = new NearCacheConfig()
  .setName("myDataStructure")
  .setInMemoryFormat(InMemoryFormat.BINARY)
  .setSerializeKeys(true)
  .setInvalidateOnChange(false)
  .setTimeToLiveSeconds(1)
  .setMaxIdleSeconds(5)
  .setEvictionConfig(evictionConfig)
  .setCacheLocalEntries(true)
  .setLocalUpdatePolicy(NearCacheConfig.LocalUpdatePolicy.CACHE_ON_UPDATE)
  .setPreloaderConfig(preloaderConfig);

setInvalidateOnChange metodu

TRUE ise altta kullanılan veri yapısı değişirse NearCache veri yapısı da değiştirilir.

Örnek

Şöyle yaparız

EvictionConfig evictionConfig = new EvictionConfig()
  .setEvictionPolicy(EvictionPolicy.NONE)
  .setSize(1_000_000)
  .setMaxSizePolicy(MaxSizePolicy.ENTRY_COUNT);

NearCacheConfig nearCacheConfig = new NearCacheConfig()
  .setInMemoryFormat(InMemoryFormat.OBJECT)
  .setInvalidateOnChange(true)
  .setSerializeKeys(true)
  .setEvictionConfig(evictionConfig)
  .setCacheLocalEntries(true);

setMaxIdleSeconds metodu

HazelcastAPI Max Idle Seconds ve TTL yazısına bakabilirsiniz

setTimeToLiveSeconds metodu

HazelcastAPI Max Idle Seconds ve TTL yazısına bakabilirsiniz

Örnek

Şöyle yaparız

ClientConfig createClientConfig() {
  ClientConfig clientConfig = new ClientConfig();
  clientConfig.addNearCacheConfig(createNearCacheConfig());
  return clientConfig;
}

NearCacheConfig createNearCacheConfig() {
  NearCacheConfig nearCacheConfig = new NearCacheConfig();
  nearCacheConfig.setName("cars");
  nearCacheConfig.setTimeToLiveSeconds(360);
  nearCacheConfig.setMaxIdleSeconds(60);
  return nearCacheConfig;
}

Hazelcast Jet JetService Arayüzü

Giriş 

Şu satırı dahil ederiz

import com.hazelcast.jet.JetService;

Kalıtım şöyle

JetService 

  JetInstance

    AbstractJetInstance

      JetInstanceImpl : Member-side JetInstance implementation

      JetClientInstanceImpl : Client-side JetInstance implementation

      BootstrappedJetProxy

getJob metodu - String

İsmi belirtilen iş varsa Job döndürür.

new Job(...) ve newLightJob(...) çağrıları arasında fark var. Açıklaması şöyle

Regular jobs 

created via JetService#newJob(..). These use MasterContext

Light jobs 

created via JetService#newLightJob(..). These use LightMasterContext. Light jobs less overhead but have some limitations (no custom resources upload) and they are used for all SQL queries.

newJob metodu - Pipeline + JobConfig

JobConfig nesnes alır

Örnek

Şöyle yaparız

HazelcastInstance hazelcastInstance = ...
        
Pipeline pipeline = Pipeline.create()
  .readFrom(TestSources.items("22578"))
  .writeTo(Sinks.logger())
  .getPipeline();

JobConfig jobConfig = new JobConfig();
jobConfig.setName("job22578");
        
TimeUnit.SECONDS.sleep(5L);
System.out.println("Launching " + jobConfig.getName());

Job job = hazelcastInstance.getJet().newJob(pipeline, jobConfig);

System.out.println("Launched " + jobConfig.getName() + "," + job.getStatus() + "," + job);
TimeUnit.HOURS.sleep(1L);
        
hazelcastInstance.shutdown();

Hazelcast Jet TaskletExecutionService Sınıfı

Giriş

Şu satırı dahil ederiz

import com.hazelcast.jet.impl.execution.TaskletExecutionService;

Tasklet'leri çalıştıran şey TaskletExecutionService. Job'ı temsil eden şey ise LightMasterContext

Kod şöyle

public class TaskletExecutionService {

  public static final String TASKLET_INIT_CLOSE_EXECUTOR_NAME = "jet:tasklet_initClose";

  private final ExecutorService blockingTaskletExecutor = 
    newCachedThreadPool(new BlockingTaskThreadFactory());
  private final ExecutionService hzExecutionService;
  private final CooperativeWorker[] cooperativeWorkers;
  private final Thread[] cooperativeThreadPool;
  private final String hzInstanceName;
  private final ILogger logger;
  private int cooperativeThreadIndex;
  Probe(name = "blockingWorkerCount")
  private final Counter blockingWorkerCount = MwCounter.newMwCounter();
  private volatile boolean isShutdown;
  private final Object lock = new Object();
  private final IdleStrategy idlerCooperative;
  private final IdleStrategy idlerNonCooperative;
  ...
}

Tasklet Nedir?

Stage'i kodda temsil eden şey. Tasklet'in bir girdi Inbox'ı bir de çıktı Outbox'ı vardır. Açıklaması şöyle

Hazelcast creates multiple parallel tasklets for each stage. It transfers the data between the tasklets of consecutive stages using two main routing strategies:

round-robin: a load-balancing edge that sends items to tasklets in a round-robin fashion. If a given queue is full, it tries the next one.

isolated: isolates the parallel code paths from each other, thereby preserving the order of events in each path. When the two connected vertices have the same parallelism, it establishes one-to-one connections between tasklets.

partitioned: computes the partition key of every item, which uniquely determines the destination tasklet. Necessary for stateful keyed transformations like group-and-aggregate.

Round-robin is the default strategy. This means that an event emitted by a tasklet can be routed to any tasklet of the following stage. This strategy results in good balancing of the load of every CPU core, but it introduces event reordering.

İskeleti şöyle

interface Tasklet {
   ProgressState call();
   ...
}

Açıklaması şöyle

The execution engine repeatedly invokes call(), which is supposed to return in no more than 1 millisecond. The ProgressState result is a pair of booleans: (madeProgress, isDone). The former is used for CPU load control (to prevent a hot idle loop) and the latter signals the completion of the task. As long as the task keeps reporting "not done", Jet will call it again.

Tasklet Processor'ı Çağırır

Processor şöyle

interface Processor {
   void process(int ordinal, Inbox inbox);
   ...
}

Açıklaması şöyle

ordinal identifies the input edge and inbox contains a batch of input data. The tasklet keeps calling this method until it has consumed all the items from the inbox and then refills the inbox with more data (possibly from a different input edge).

CooperativeWorker Sınıfı

TaskletExecutionService içindeki bir sınıf

run metodu

Metod kabaca şöyle

private final class CooperativeWorker implements Runnable {
  ...
  @Override
  public void run() {
    while (true) {
      boolean madeProgress = false;
      long idleCount = 0;
      for (Tasklet t : tasklets) {
        // let one tasklet do a piece of work       
        madeProgress |= t.call();
      }
      if (madeProgress) {
        idleCount = 0;
      } else {
        idle(idleCount++);
      }
    }
  }
}

Açıklaması şöyle

The worker repeatedly calls the tasklets’ call() method. The method returns whether the tasklet made any progress. In case of processors, it returns true when the processor took anything from the inbox or put anything to the outbox (and few more special cases).

The idle() call is a brief sleep in case none of the workers made progress. In Jet 3.0 it’s between 25 and 1000µs, depending on the idleCount (*2). As you probably can see now, if just one of the tasklets made progress, there won’t be any sleeping at all and we’ll proceed to the next iteration immediately. What this can cause is that even with pretty low traffic, the CPU usage will be quite high, especially if there are many tasklets per worker.

Hazelcast SQL - Table-Valued Functions

Giriş

Açıklaması şöyle

Table-valued functions return tables of batch or streaming sources that you can use in SQL statements.

GENERATE_SERIES

Açıklaması şöyle

Returns a table that contains a series of numbers, starting from the start_number argument and ending with the stop_number arguments.

Örnek

Şöyle yaparız. 100 milyon satırı IMap'e yazar

CREATE MAPPING my_first_map 
  TYPE IMap 
  OPTIONS ('keyFormat'='varchar','valueFormat'='varchar');

INSERT INTO my_first_map
  SELECT CAST(v AS VARCHAR), CAST(v AS VARCHAR) 
    FROM (SELECT * FROM TABLE(generate_series(0,100000000)))

Örnek

Şöyle yaparız. v isimli bir sütun verir

SELECT * FROM TABLE(generate_series(1,3));
+------------+
|           v|
+------------+
|           1|
|           2|
|           3|
+------------+

Örnek

Şöyle yaparız. Burada key değeri birden başlıyor yüz bine kadar. value değeri hep sıfır

sql> CREATE MAPPING test_map TYPE IMap OPTIONS ('keyFormat'='int', 'valueFormat'='int');
OK
sql> insert into test_map select v, 0 from table(generate_series(1, 100000));
OK
sql> select count(*) from test_map;
+--------------------+
|              EXPR$0|
+--------------------+
|              100000|
+--------------------+
1 row(s) selected

GENERATE_STREAMS

Açıklaması şöyle

Returns a table that contains a stream of numbers, starting from 0 at a rate of numbers_per_second every second, without an upper bound

Örnek

Şöyle yaparız

SqlService sql = ...;
try (SqlResult result = sql.execute("SELECT * FROM TABLE(GENERATE_STREAM(5)) LIMIT 2")) {
  Iterator<SqlRow> rows = result.iterator();
  while (rows.hasNext()) {
    System.out.println("row :" + rows.next());
  }
}

Örnek

Şöyle yaparız. Burada CancellationException alırız

HazelcastInstance hz = ...

SqlResult result = hz.getSql().execute("SELECT * FROM TABLE(generate_stream(1))");
result.iterator().next();
result.close();

Hazelcast SQL - JSON Functions

Giriş

Açıklaması şöyle

Use standard JSON SQL functions for OBJECT and ARRAY aggregations

Bunlar şöyle

JSON_QUERY

JSON_VALUE

JSON_ARRAY

JSON_OBJECT

1. JSON_QUERY

Açıklaması şöyle

The JSON_QUERY() function extracts a JSON value from a JSON document or a JSON-formatted string that matches a given JsonPath expression.

4 parametre alıyor. wrapperBehavior ve onClauseArg isteğe bağlı

jsonArg: JSON value or a JSON-formatted string.

jsonPath: A JsonPath expression that identifies the data that you want to get from the jsonArg parameter.

wrapperBehavior: What to do with return results.

  WITHOUT [ARRAY] WRAPPER (default): Up to one matched value is returned without wrapping in an array. Fails, if multiple values match.

  WITH [CONDITIONAL] [ARRAY] WRAPPER: Returns a single match directly without wrapping. If there are multiple matches, they are returned as a JSON array.

  WITH UNCONDITIONAL [ARRAY] WRAPPER: Always wrap matched values in a JSON array.

onClauseArg: A value to return in case of the given ON condition:

 EMPTY ARRAY

 EMPTY OBJECT

 ERROR

 NULL

Örnek

Elimizde şöyle bir JSON olsun

'{
    "CustomerId": "123456",
    "Orders": [{
            "OrderId": "852",
            "OrderManifests": [{
                    "ShippedProductId": 884,
                    "ProductId": 884
                }, {
                    "ShippedProductId": 951,
                    "ProductId": 2564
                }    
            ]
        }, {
            "OrderId": "5681",
            "OrderManifests": [{
                    "ShippedProductId": 198,
                    "ProductId": 4681
                }, {
                    "ShippedProductId": 8188,
                    "ProductId": 8188
                }, {
                    "ShippedProductId": 144,
                    "ProductId": 8487
                }
            ]
        }
    ]
}'

OrderManifest'leri almak için şöyle yaparız

"SELECT JSON_QUERY(this, '$..OrderManifests[*]' WITH ARRAY WRAPPER) 
  as OrderManifests FROM "  + mapName

OrderManifest'leri müşteri numarasına göre almak için şöyle yaparız

SELECT JSON_QUERY(this, '$..OrderManifests[*]' WITH ARRAY WRAPPER) 
  as OrderManifests FROM " + mapName 
  + " WHERE JSON_VALUE(this, '$.CustomerId' returning BIGINT) = customerID

müşteri numarası string ise şöyle yaparız

SELECT JSON_QUERY(this, '$..OrderManifests[*]' WITH ARRAY WRAPPER) 
  as OrderManifests FROM " + mapName 
  + " WHERE JSON_VALUE(this, '$.CustomerId' returning VARCHAR) = 'customerID'

2. JSON_VALUE

Gerçekleştirimi com.hazelcast.jet.sql.impl.expression.json.JsonValueFunction sınıfında

Açıklaması şöyle

The JSON_VALUE() function extracts a primitive value, such as a string, number, or boolean that matches a given JsonPath expression. This function returns NULL if a non-primitive value is matched, unless the ON ERROR behavior is changed.

4 parametre alıyor. RETURNING ve onClauseArg isteğe bağlı

jsonArg: JSON value or a JSON-formatted string

jsonPath: A JsonPath expression that identifies the data that you want to get from the jsonArg parameter.

RETURNING: Converts the result to the dataType (VARCHAR by default). If the value cannot be converted to the target type, throws an error.

onClauseArg: What to return in case of the ON condition:

  DEFAULT <literal | column | parameter>

  ERROR

  NULL

Örnek

Şöyle yaparız

IMap<Integer, HazelcastJsonValue> map = instance().getMap(mapName);
map.put(42, new HazelcastJsonValue("{\"first_name\":\"value\"}"));
SqlResult rows = instance().getSql().execute("select * from " + mapName + 
  " where JSON_VALUE(this, '$.\"first_name\"' returning VARCHAR) like 'v%'");

hz-cli komutu - Utility to perform operations on a Hazelcast cluster

Giriş

Bu komut aslında bir shell script. Kodu şöyle. Yani altta Picocli kullanan bir java kodunu çalıştırıyor. $@ ile hz-cli script'ine geçilen tüm parametreler HazelcastCommandLine sınıfına geçiliyor

#!/bin/bash

...

JAVA_OPTS_ARRAY=(\
$JDK_OPTS \
"-Dhazelcast.client.config=$HAZELCAST_CLIENT_CONFIG" \
$JVM_OPTIONS \
$JAVA_OPTS \
)

$JAVA "${JAVA_OPTS_ARRAY[@]}" -\
  cp "$CLASSPATH" com.hazelcast.client.console.HazelcastCommandLine "$@"

Global seçenekler şöyle. Global seçenekler komutlardan önce gelmelidir

--ignore-version-mismatch

Ignore check between client and server versions (by default they must have matching minor version)

-t, --targets=[<cluster-name>@]<hostname>:<port>[,<hostname>:<port>]

The cluster name and addresses to use if you want to connect to a cluster other than the one configured in the configuration file. At least one address is required. The cluster name is optional.

-h, --help              

Show this help message and exit.

-V, --version           

Print version information and exit.

-f, --config=<config>   

Optional path to a client config XML/YAML file. The default is to use config/hazelcast-client.xml.

-v, --verbosity         

Show verbose logs and full stack trace of errors

-v seçeneği verbose anlamına gelir. Data detaylı çıktı verir

Komutlar şöyle

help             Displays help information about the specified command

cancel         Cancels a running job

cluster         Shows current cluster state and information about members

console       Starts the console application for trying out in-memory data

                     structures of Hazelcast. It is not recommended for use in

                     production.

delete-snapshot  Deletes a named snapshot

list-jobs       Lists running jobs on the cluster

list-snapshots   Lists exported snapshots on the cluster

restart         Restarts a running job

resume        Resumes a suspended job

save-snapshot    Exports a named snapshot from a job and optionally cancels it

sql               Starts the SQL shell [BETA]

submit         Submits a job to the cluster

suspend      Suspends a running job

cancel komutu

Örnek

Şöyle yaparız

> bin/hz-cli submit target/pulsar-example-1.0-SNAPSHOT.jar
> bin/hz-cli cancel pulsar-message-counter
> hz-stop

Örnek

Şöyle yaparız

> hz-cli cancel wonky

cluster komutu

Örnek

Şöyle yaparız

$ hz-cli cluster
State: ACTIVE
Version: 5.3.2
Size: 3

ADDRESS                  UUID               
[172.26.0.4]:5701        4c810f68-cadd-42d5-8bd1-8028abc1d389
[172.26.0.3]:5701        4d648b81-a541-432f-b2fe-241f84bb4c87
[172.26.0.2]:5701        68774596-6c1b-4246-9755-7bfdd09b60f9

console komutu

Console App uygulamasını başlatır

help komutu

Global seçenekleri gösterir. Çıktısı şöyle.

$ bin/hz-cli --help
Hazelcast 5.2.0
Usage: hz-cli [-hvV] [--ignore-version-mismatch] [-f=<config>] [-t=
              [<cluster-name>@]<hostname>:<port>[,<hostname>:<port>]] [COMMAND]
Utility to perform operations on a Hazelcast cluster.
By default it uses the file config/hazelcast-client.xml to configure the client
connection.

Global options are:

  --ignore-version-mismatch
                          Ignore check between client and server versions (by
                          default they must have matching minor version)
  -t, --targets=[<cluster-name>@]<hostname>:<port>[,<hostname>:<port>]
                          The cluster name and addresses to use if you want to
                          connect to a cluster other than the one configured
                          in the configuration file. At least one address is
                          required. The cluster name is optional.
  -h, --help              Show this help message and exit.
  -V, --version           Print version information and exit.
  -f, --config=<config>   Optional path to a client config XML/YAML file. The
                          default is to use config/hazelcast-client.xml.
  -v, --verbosity         Show verbose logs and full stack trace of errors

list jobs komutu

Örnek

Şöyle yaparız

hz-cli -t $MEMBER_IP list-jobs

Örnek

Şöyle yaparız

> hz-cli list-jobs
ID         		  STATUS       SUBMISSION TIME     	   NAME
097e-6a3d-1e00-0001 RUNNING      2023-03-03T13:13:09.390 N/A

submit komutu

Örnek - n

Çıktısı şöyle

> hz-cli submit -n foo basic-hazelcast-jet-1.0-SNAPSHOT.jar
Submitting JAR 'basic-hazelcast-jet-1.0-SNAPSHOT.jar' with arguments []
Using job name 'foo'
The JAR didn't submit any jobs.

Örnek -- help

Çıktısı şöyle

$ hz-cli submit --help

Usage:hz-cli submit [-v] [--ignore-version-mismatch] [-c=] [-n=] [-s=] [-t=[@]:[,:]] [...]


Submits a job to the cluster
The jar file to submit

[...] 
Arguments to pass to the supplied jar file

-c, --class= 
Fully qualified name of the main class inside the JAR file

--ignore-version-mismatch
Ignore check between client and server versions (by
default they must have matching minor version)

-n, --name= 
Name of the job

-s, --snapshot= 
Name of the initial snapshot to start the job from

-t, --targets=[@]:[,:] 
The cluster name and addresses to use if you want to
connect to a cluster other than the one configured
in the configuration file. At least one address is
required. The cluster name is optional.

-v, --verbosity Show verbose logs and full stack trace of errors

Aslında belirtilen jar'ın main sınıfını çalıştırır. Main sınıf' şöyle olduğu için smart olmayan bir tane Hazelcast client başlatır ve onu kullanarak job'ı yükler.

public static void main(String[] args) {

  Pipeline pipeline = ...

  HazelcastInstance hz = Hazelcast.bootstrappedInstance(); 

  hz.getJet().newJob(pipeline); 
}

Örnek

Şöyle yaparız. Burada MANIFEST.MF dosyasında Main-Class alanına değer atandığı varsayılıyor. Böylece jar çalıştırılırken hangi class'tan başlayacağını belirtmeye gerek yok. Eğer bu alan tanımlı değilse --class seçeneği ile main sınıfı belirtmek gerekir.

$ hz-cli submit myfat.jar

Örnek - f seçeneği

Şöyle yaparız. Burada belirtilen xml kullanılıyor.

$ bin/hz-cli -f hazelcast-client.xml submit hello-world.jar
Submitting JAR 'hello-world.jar' with arguments []

Örnek --class seçeneği

MANIFEST.MF dosyasında Main-Class alanına değer yoksa şöyle yaparız

$ hz-cli submit --class org.example.EvenNumberStream $PATH_TO_JAR_FILE

sql komutu

Bir SQL kabuğu (shell) başlatır. Bu kabuk içinden SQL cümleleri çalıştırılabilir

Örnek

Şöyle yaparız

sql> CREATE MAPPING people DATA CONNECTION "mysql";

Örnek

Şöyle yaparız

sql> CREATE OR REPLACE MAPPING mymap (__key INT, this VARCHAR) 
TYPE IMap 
OPTIONS ('keyFormat'='int','valueFormat'='varchar');

SELECT * FROM mymap;