Szybki start: odbieranie zdarzeń z usługi Event Hubs przy użyciu systemu Apache Storm
Apache Storm to rozproszony system obliczeniowy w czasie rzeczywistym, który upraszcza niezawodne przetwarzanie niezwiązanych strumieni danych. W tej sekcji pokazano, jak używać spout Azure Event Hubs Storm do odbierania zdarzeń z usługi Event Hubs. Za pomocą systemu Apache Storm można podzielić zdarzenia między wiele procesów hostowanych w różnych węzłach. Integracja usługi Event Hubs z systemem Storm upraszcza zużycie zdarzeń przez przezroczyste tworzenie punktów kontrolnych postępu przy użyciu instalacji usługi Zookeeper systemu Storm, zarządzanie trwałymi punktami kontrolnymi i równoległymi odbiorami z usługi Event Hubs.
Aby uzyskać więcej informacji na temat wzorców odbierania usługi Event Hubs, zobacz Omówienie usługi Event Hubs.
Wymagania wstępne
Przed rozpoczęciem pracy z przewodnikiem Szybki start utwórz przestrzeń nazw usługi Event Hubs i centrum zdarzeń. Użyj Azure Portal, aby utworzyć przestrzeń nazw typu Event Hubs i uzyskać poświadczenia zarządzania wymagane przez aplikację do komunikowania się z centrum zdarzeń. Aby utworzyć przestrzeń nazw i centrum zdarzeń, wykonaj procedurę opisaną w tym artykule.
Tworzenie projektu i dodawanie kodu
Użyj następującego polecenia, aby zainstalować pakiet w lokalnym magazynie Maven. Dzięki temu można dodać go jako odwołanie w projekcie Storm w późniejszym kroku.
mvn install:install-file -Dfile=target\eventhubs-storm-spout-0.9-jar-with-dependencies.jar -DgroupId=com.microsoft.eventhubs -DartifactId=eventhubs-storm-spout -Dversion=0.9 -Dpackaging=jar
W środowisku Eclipse utwórz nowy projekt Maven (kliknij pozycję Plik, Nowy, Projekt).
Wybierz pozycję Użyj domyślnej lokalizacji obszaru roboczego, a następnie kliknij przycisk Dalej
Wybierz archetyp maven-archetype-quickstart , a następnie kliknij przycisk Dalej
Wstaw wartości GroupId i ArtifactId, a następnie kliknij przycisk Zakończ
W pom.xmldodaj następujące zależności w węźle
<dependency>
.<dependency> <groupId>org.apache.storm</groupId> <artifactId>storm-core</artifactId> <version>0.9.2-incubating</version> <scope>provided</scope> </dependency> <dependency> <groupId>com.microsoft.eventhubs</groupId> <artifactId>eventhubs-storm-spout</artifactId> <version>0.9</version> </dependency> <dependency> <groupId>com.netflix.curator</groupId> <artifactId>curator-framework</artifactId> <version>1.3.3</version> <exclusions> <exclusion> <groupId>log4j</groupId> <artifactId>log4j</artifactId> </exclusion> <exclusion> <groupId>org.slf4j</groupId> <artifactId>slf4j-log4j12</artifactId> </exclusion> </exclusions> <scope>provided</scope> </dependency>
W folderze src utwórz plik o nazwie Config.properties i skopiuj następującą zawartość, podstawiając
receive rule key
wartości ievent hub name
:eventhubspout.username = ReceiveRule eventhubspout.password = {receive rule key} eventhubspout.namespace = ioteventhub-ns eventhubspout.entitypath = {event hub name} eventhubspout.partitions.count = 16 # if not provided, will use storm's zookeeper settings # zookeeper.connectionstring=localhost:2181 eventhubspout.checkpoint.interval = 10 eventhub.receiver.credits = 10
Wartość eventhub.receiver.credit określa, ile zdarzeń jest wsadowych przed ich zwolnieniem do potoku Storm. Dla uproszczenia ten przykład ustawia tę wartość na 10. W środowisku produkcyjnym zwykle należy ustawić je na wyższe wartości; na przykład 1024. 1 . Utwórz nową klasę o nazwie LoggerBolt przy użyciu następującego kodu:
import java.util.Map; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import backtype.storm.task.OutputCollector; import backtype.storm.task.TopologyContext; import backtype.storm.topology.OutputFieldsDeclarer; import backtype.storm.topology.base.BaseRichBolt; import backtype.storm.tuple.Tuple; public class LoggerBolt extends BaseRichBolt { private OutputCollector collector; private static final Logger logger = LoggerFactory .getLogger(LoggerBolt.class); @Override public void execute(Tuple tuple) { String value = tuple.getString(0); logger.info("Tuple value: " + value); collector.ack(tuple); } @Override public void prepare(Map map, TopologyContext context, OutputCollector collector) { this.collector = collector; this.count = 0; } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { // no output fields } }
Ten element Bolt storm rejestruje zawartość odebranych zdarzeń. Można to łatwo rozszerzyć, aby przechowywać krotki w usłudze magazynu. Przykład usługi HDInsight Storm z usługą Event Hub używa tego samego podejścia do przechowywania danych w usługach Azure Storage i Power BI.
Utwórz klasę o nazwie LogTopology z następującym kodem:
import java.io.FileReader; import java.util.Properties; import backtype.storm.Config; import backtype.storm.LocalCluster; import backtype.storm.StormSubmitter; import backtype.storm.generated.StormTopology; import backtype.storm.topology.TopologyBuilder; import com.microsoft.eventhubs.samples.EventCount; import com.microsoft.eventhubs.spout.EventHubSpout; import com.microsoft.eventhubs.spout.EventHubSpoutConfig; public class LogTopology { protected EventHubSpoutConfig spoutConfig; protected int numWorkers; protected void readEHConfig(String[] args) throws Exception { Properties properties = new Properties(); if (args.length > 1) { properties.load(new FileReader(args[1])); } else { properties.load(EventCount.class.getClassLoader() .getResourceAsStream("Config.properties")); } String username = properties.getProperty("eventhubspout.username"); String password = properties.getProperty("eventhubspout.password"); String namespaceName = properties .getProperty("eventhubspout.namespace"); String entityPath = properties.getProperty("eventhubspout.entitypath"); String zkEndpointAddress = properties .getProperty("zookeeper.connectionstring"); // opt int partitionCount = Integer.parseInt(properties .getProperty("eventhubspout.partitions.count")); int checkpointIntervalInSeconds = Integer.parseInt(properties .getProperty("eventhubspout.checkpoint.interval")); int receiverCredits = Integer.parseInt(properties .getProperty("eventhub.receiver.credits")); // prefetch count // (opt) System.out.println("Eventhub spout config: "); System.out.println(" partition count: " + partitionCount); System.out.println(" checkpoint interval: " + checkpointIntervalInSeconds); System.out.println(" receiver credits: " + receiverCredits); spoutConfig = new EventHubSpoutConfig(username, password, namespaceName, entityPath, partitionCount, zkEndpointAddress, checkpointIntervalInSeconds, receiverCredits); // set the number of workers to be the same as partition number. // the idea is to have a spout and a logger bolt co-exist in one // worker to avoid shuffling messages across workers in storm cluster. numWorkers = spoutConfig.getPartitionCount(); if (args.length > 0) { // set topology name so that sample Trident topology can use it as // stream name. spoutConfig.setTopologyName(args[0]); } } protected StormTopology buildTopology() { TopologyBuilder topologyBuilder = new TopologyBuilder(); EventHubSpout eventHubSpout = new EventHubSpout(spoutConfig); topologyBuilder.setSpout("EventHubsSpout", eventHubSpout, spoutConfig.getPartitionCount()).setNumTasks( spoutConfig.getPartitionCount()); topologyBuilder .setBolt("LoggerBolt", new LoggerBolt(), spoutConfig.getPartitionCount()) .localOrShuffleGrouping("EventHubsSpout") .setNumTasks(spoutConfig.getPartitionCount()); return topologyBuilder.createTopology(); } protected void runScenario(String[] args) throws Exception { boolean runLocal = true; readEHConfig(args); StormTopology topology = buildTopology(); Config config = new Config(); config.setDebug(false); if (runLocal) { config.setMaxTaskParallelism(2); LocalCluster localCluster = new LocalCluster(); localCluster.submitTopology("test", config, topology); Thread.sleep(5000000); localCluster.shutdown(); } else { config.setNumWorkers(numWorkers); StormSubmitter.submitTopology(args[0], config, topology); } } public static void main(String[] args) throws Exception { LogTopology topology = new LogTopology(); topology.runScenario(args); } }
Ta klasa tworzy nowy element spout usługi Event Hubs przy użyciu właściwości w pliku konfiguracji w celu utworzenia wystąpienia. Należy pamiętać, że w tym przykładzie jest tworzonych tyle zadań spouts jak liczba partycji w centrum zdarzeń w celu użycia maksymalnej równoległości dozwolonej przez to centrum zdarzeń.
Następne kroki
Następujące linki pozwalają dowiedzieć się więcej na temat usługi Event Hubs: