streamalism.org

1. Inleiding

In deze tutorial leren we over Hazelcast Jet. Het is een gedistribueerde gegevensverwerkingsengine van Hazelcast, Inc. en is bovenop Hazelcast IMDG gebouwd.

Als u meer wilt weten over Hazelcast IMDG, vindt u hier een artikel om aan de slag te gaan.

2. Wat is Hazelcast Jet?

Hazelcast Jet is een gedistribueerde gegevensverwerkingsengine die gegevens als streams behandelt. Het kan gegevens verwerken die zijn opgeslagen in een database of bestanden, evenals de gegevens die worden gestreamd door een Kafka-server.

Bovendien kan het geaggregeerde functies uitvoeren over oneindige datastromen door de streams in subsets te verdelen en aggregatie over elke subset toe te passen. Dit concept staat bekend als windowing in de Jet-terminologie.

We kunnen Jet inzetten in een cluster van machines en vervolgens onze gegevensverwerkingstaken eraan bezorgen. Jet zorgt ervoor dat alle leden van het cluster de gegevens automatisch verwerken. Elk lid van het cluster verbruikt een deel van de gegevens, en dat maakt het eenvoudig om op te schalen naar elk niveau van doorvoer.

Hier zijn de typische gebruiksscenario's voor Hazelcast Jet:

• Real-time stroomverwerking
• Snelle batchverwerking
• Verwerking van Java 8 Streams op een gedistribueerde manier
• Gegevensverwerking in Microservices

3. Installatie

Om Hazelcast Jet in onze omgeving in te stellen, hoeven we slechts een enkele Maven-afhankelijkheid toe te voegen aan onze pom.xml.

Hier is hoe we het doen:

 com.hazelcast.jet hazelcast-jet 4.2 

Met deze afhankelijkheid wordt een jar-bestand van 10 Mb gedownload dat ons alle infrastructuur biedt die we nodig hebben om een ​​gedistribueerde pijplijn voor gegevensverwerking te bouwen.

De nieuwste versie voor Hazelcast Jet is hier te vinden.

4. Voorbeeldtoepassing

Om meer te weten te komen over Hazelcast Jet, zullen we een voorbeeldtoepassing maken die een invoer van zinnen en een woord nodig heeft om in die zinnen te zoeken en het aantal opgegeven woorden in die zinnen retourneert.

4.1. De pijpleiding

Een Pipeline vormt de basisconstructie voor een Jet-applicatie. Verwerking binnen een pijplijn volgt deze stappen:

• gegevens uit een bron lezen
• transformeer de gegevens
• schrijf gegevens in een sink

Voor onze applicatie leest de pijplijn uit een gedistribueerde Lijst, pas de transformatie van groepering en aggregatie toe en schrijf ten slotte naar een gedistribueerd Kaart.

Hier is hoe we onze pijplijn schrijven:

private Pipeline createPipeLine () {Pipeline p = Pipeline.create (); p.readFrom (Sources.list (LIST_NAME)) .flatMap (word -> traverseArray (word.toLowerCase (). split ("\ W +"))) .filter (word ->! word.isEmpty ()) .groupingKey (wholeItem ()) .aggregate (counting ()) .writeTo (Sinks.map (MAP_NAME)); terug p; }

Nadat we van de bron hebben gelezen, doorlopen we de gegevens en splitsen we deze door de ruimte met behulp van een reguliere expressie. Daarna filteren we de lege plekken eruit.

Ten slotte groeperen we de woorden, voegen ze samen en schrijven de resultaten naar een Kaart.

4.2. De baan

Nu onze pijplijn is gedefinieerd, creëren we een taak voor het uitvoeren van de pijplijn.

Hier is hoe we een countWord functie die parameters accepteert en de telling retourneert:

openbaar Long countWord (Lijstzinnen, String-woord) {long count = 0; JetInstance jet = Jet.newJetInstance (); probeer {List textList = jet.getList (LIST_NAME); textList.addAll (zinnen); Pijplijn p = createPipeLine (); jet.newJob (p) .join (); Kaartaantallen = jet.getMap (MAP_NAME); count = counts.get (woord); } ten slotte {Jet.shutdownAll (); } aantal keren terug; }

We maken eerst een Jet-instantie om onze taak te creëren en de pijplijn te gebruiken. Vervolgens kopiëren we de invoer Lijst naar een gedistribueerde lijst, zodat deze beschikbaar is voor alle instanties.

Vervolgens dienen we een opdracht in via de pijpleiding die we hierboven hebben aangelegd. De methode nieuwe baan() geeft een uitvoerbare taak terug die asynchroon is gestart door Jet. De toetreden methode wacht tot de taak is voltooid en gooit een uitzondering als de taak is voltooid met een fout.

Wanneer de taak is voltooid, worden de resultaten opgehaald in een gedistribueerde Kaart, zoals we in onze pijplijn hebben gedefinieerd. Dus we krijgen de Kaart van de Jet-instantie en haal de tellingen van het woord ertegen op.

Ten slotte sluiten we de Jet-instantie af. Het is belangrijk om het af te sluiten nadat onze uitvoering is beëindigd, zoals Jet-instantie begint zijn eigen threads. Anders blijft ons Java-proces actief, zelfs nadat onze methode is afgesloten.

Hier is een unit-test die de code test die we voor Jet hebben geschreven:

@Test openbare leegte whenGivenSentencesAndWord_ThenReturnCountOfWord () {Lijst zinnen = nieuwe ArrayList (); zinnen.add ("De eerste seconde was in orde, maar de tweede was moeilijk."); WordCounter wordCounter = nieuwe WordCounter (); long countSecond = wordCounter.countWord (zinnen, "tweede"); assertEquals (3, countSecond); }

5. Conclusie

In dit artikel hebben we geleerd over Hazelcast Jet. Raadpleeg de handleiding voor meer informatie over het en zijn functies.

Zoals gewoonlijk is de code voor de voorbeelden die in dit artikel worden gebruikt, te vinden op Github.