Open Source Storage and Computing at In-Memory Speeds

// Start the Embedded Hazelcast Cluster Member.
HazelcastInstance hz = Hazelcast.newHazelcastInstance();
 // Get the Distributed Map from Cluster.
IMap map = hz.getMap("my-distributed-map");
//Standard Put and Get.
map.put("key", "value");
map.get("key");
//Concurrent Map methods, optimistic updating
map.putIfAbsent("somekey", "somevalue");
map.replace("key", "value", "newvalue");
// Shutdown the Hazelcast Cluster Member
hz.shutdown();

IMap<String,Double> averagePrices = jet.getMap("current-avg-trade-price");

Pipeline p = Pipeline.create();
// Stream (trade symbol, price) records from Kafka.
p.readFrom(KafkaSources.kafka("trades", props))
 .withTimestamps(Trade::getTime, 0L)
 .filter(trade -> STOCKS.contains(trade.getSymbol()))
 .groupingKey(Trade::getSymbol)
 // 10 second sliding window, updated every 100ms
 .window(WindowDefinition.sliding(10_000, 100))
 .aggregate(AggregateOperations.averagingLong(Trade::getPrice)) 
 // write results to a distributed map.
 .map(window -> Util.entry(window.getKey(), window.getValue()))
 .writeTo(Sinks.map(averagePrices));

<map name="customers">
    <backup-count>1</backup-count>
    <eviction eviction-policy="NONE" 
              max-size-policy="PER_NODE" size="0"/>
    <map-store enabled="true" initial-mode="LAZY">
        <class-name>com.examples.DummyStore</class-name>
        <write-delay-seconds>60</write-delay-seconds>
        <write-batch-size>1000</write-batch-size>
        <write-coalescing>true</write-coalescing>
        <properties>
           <property name="jdbc_url">my.jdbc.com</property>
        </properties>
    </map-store>
</map-store>

IMap<Integer, Product> productMap = jet.getMap(PRODUCTS);
Pipeline p = Pipeline.create();
// read a list of trades, join with the product map and then write back to files
p.readFrom(Sources.files("trades"))
 .mapUsingIMap(productMap, 
       trade -> trade.productId(), 
       (t, product) -> tuple2(t, product.name())
 ).writeTo(Sinks.files("joined"));