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 nieprzewią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 rozwiązaniem Storm upraszcza zużycie zdarzeń przez przezroczyste określanie postępu przy użyciu instalacji usługi Zookeeper systemu Storm, zarządzania 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, które aplikacja musi komunikować się z centrum zdarzeń. Aby utworzyć przestrzeń nazw i centrum zdarzeń, wykonaj procedurę opisaną w tym artykule.

Tworzenie projektu i dodawanie kodu

  1. Użyj następującego polecenia, aby zainstalować pakiet w lokalnym magazynie Maven. Dzięki temu można dodać go jako odwołanie do projektu 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
    
  2. W środowisku Eclipse utwórz nowy projekt Maven (kliknij pozycję Plik, a następnie pozycję Nowy, a następnie Projekt).

    Plik — nowy —>> projekt

  3. Wybierz pozycję Użyj domyślnej lokalizacji obszaru roboczego, a następnie kliknij przycisk Dalej

  4. Wybierz archetyp archetypu maven-archetype-quickstart , a następnie kliknij przycisk Dalej

  5. Wstaw identyfikator GroupId i ArtifactId, a następnie kliknij przycisk Zakończ

  6. 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>
    
  7. W folderze src utwórz plik o nazwie Config.properties i skopiuj następującą zawartość, zastępując receive rule key wartości i event 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ść parametru eventhub.receiver.credit określa, ile zdarzeń jest wsadowych przed zwolnieniem ich do potoku Storm. Dla uproszczenia ten przykład ustawia tę wartość na 10. W środowisku produkcyjnym zwykle należy ustawić na wyższe wartości; na przykład 1024. 1 . Utwórz nową klasę o nazwie LoggerBolt z następującym kodem:

     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 Storm bolt rejestruje zawartość odebranych zdarzeń. Można to łatwo rozszerzyć w celu przechowywania 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łudze Azure Storage i usłudze Power BI.

  8. 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 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 jako liczby partycji w centrum zdarzeń, aby użyć 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: