Edge Computing und Industrial IoT mit Apache Edgent und Apache PLC4X

Immer mehr vernetzte Devices produzieren immer mehr Daten. Industrie 4.0 ist in aller Munde. Nur: Wie kommen wir an die Daten im industriellen Umfeld? Wie gehen wir mit den damit einhergehenden Datenmengen um? Wie nutzen wir die beschränkten Möglichkeiten auf IoT-Devices am besten aus? Bei Apache Edgent und Apache PLC4X handelt es sich um zwei neue Projekte im Incubator der Apache Software Foundation, die es sich zum Ziel gemacht haben, genau diese Probleme zu lösen.

Immer mehr Bereiche unseres Lebens werden datenmäßig erfasst. Während wir vor zehn Jahren noch Bedenken hatten, die GPS-Funktion unseres Telefons eingeschaltet zu lassen, generieren unsere aktuellen Smartphones schier unvorstellbare Datenmengen. Fitness-Tracker erfassen unsere Vitaldaten und Bewegungsprofile. Jetzt fangen allmählich unsere Autos an, mit aktivierten Sensoren durch die Innenstädte zu fahren und senden kontinuierlich Telemetrie- und Sensordaten an die Server der Autohersteller. Schätzungen zufolge wird sich die Datenmenge, die so produziert wird, in weniger als zehn Jahren mehr als verzehnfachen. Und sicherlich werden in Zukunft „Connected Cars“ nicht nur mit den Autoherstellern kommunizieren, sondern auch untereinander.

Vor allem aber die Industrie erzeugt im Zuge der Umstellung auf „Industrie 4.0“ jeden Tag mehr Daten. An deren Beispiel möchte ich verdeutlichen, über welche Datenmengen wir hier reden. In den letzten Monaten habe ich mit vielen Vertretern der Automatisierungsindustrie gesprochen und über Optionen zur Nutzung von Open Source im Industrie-4.0-Umfeld gesprochen. Die Unternehmen, die bereits Maschinendaten sammeln und nutzen, tun dies in der Regel über sogenannte Protokolladapter oder -gateways. Diese sind notwendig, da der direkte Zugriff auf Daten der Steuerungen meist proprietär und damit schwierig ist. Entsprechend wird üblicherweise ein solcher Protokolladapter zwischen die Steuerung und das Netzwerk geklemmt. Dieses Gerät wird dann konfiguriert, um bestimmte Daten aus den Steuerungen zu lesen und in einem anderen Protokoll abrufbar zu machen. Alternativ leitet es diese direkt in einem anderen Protokoll – meist MQTT – weiter oder sendet alles direkt an die „Cloud“ des Gateway-Herstellers. In diesen Gesprächen hat sich oftmals herausgestellt, dass pro Maschine in der Regel eine mindestens vierstellige Anzahl an Datenpunkten in Intervallen von 1 bis 3 Sekunden gesammelt werden. Stellen wir uns weiter vor, dass in einem mittelgroßen Unternehmen einige Hundert dieser Maschinen im Einsatz sind.

Der Meeresspiegel steigt, auch im Data-Lake

Angenommen wir wollen moderate 2048 Werte sammeln und jeder dieser Werte ist ein einfacher 32-Bit-Integer. Dann wäre jede Abfrage 8 KB groß. Nehmen wir an, wir würden die Daten viermal pro Sekunde abfragen, dann wären das pro Sekunde ca. 32 KB. Dies summiert sich dann pro Tag auf ca. 2,7 GB. Aufs Jahr gerechnet wären das ziemlich genau 1 TB an Daten für nur eine Maschine. Um diese Datenmenge über eine mobile Datenverbindung stabil zu übertragen, müsste man schon mindestens auf eine UMTS (3G)-Verbindung zurückgreifen.

Ich denke, jeder kann sich vorstellen, dass das ein Problem werden kann. Und potenziell entstehen so auch viele Daten, die im Nachhinein gar nicht verwertet werden. Ich kann mir beispielsweise keine Auswertung vorstellen, bei der es noch 2028 wichtig ist, an welcher Position sich Motor 5 am 14.02.2018 um 10:27 Uhr befunden hat. Vielmehr würden wir uns eher für aus diesen Rohdaten abgeleitete und angereicherte Informationen interessieren: abgeschlossene Werkstücke, festgestellte Abweichungen von den Normparametern, aufgetretene Störungen, durchschnittliche Temperaturen usw.

Die bisher verfügbaren Protokolladapter sind allerdings nicht oder nur beschränkt in der Lage, eine derartige Vorverarbeitung vorzunehmen, daher werden momentan in der Regel einfach alle Rohdaten in die „Cloud“ gepumpt. Sei es nun die Private Cloud des Unternehmens oder eine kommerzielle Public Cloud. Mittels Big-Data-Clustern werden dann aus diesen Rohdaten im Nachgang die oben genannten Daten abgeleitet. Mit diesen ergibt es dann auch erst Sinn, die eigentlichen Analysen durchzuführen.

Dieses Vorgehen hat aus meiner Sicht neben dem zu übertragenden Datenvolumen, dem benötigten Speicherplatz und der benötigten Rechenzeit auch noch weitere Nachteile. Solange die Steuerung auf die direkte Bereitstellung eines Parameters ausgelegt ist, sehe ich hier weniger Probleme. Gehen wir mal davon aus, dass von einer Bestandsmaschine bestimmte Daten erhoben werden sollen, für deren Bereitstellung die Maschine oder deren Steuerung nicht ausgelegt ist. Dann wird es allerdings um einiges schwieriger.

Als simples Beispiel könnte man hier annehmen, dass Informationen über fertiggestellte Werkstücke abgeleitet werden müssen. Angenommen, man könnte diese aus der Position des Kolbens einer Presse in ebendieser Maschine ableiten. Gehen wir hier davon aus, dass ein produziertes Werkstück genau dann als produziert gilt, wenn der Kolben einer Stanze einmal an die Endposition gefahren ist und dann wieder in die Ausgangsposition zurückgekommen ist. Angenommen, die Maschine produziert nun eine Einheit pro Sekunde und die Sample-Frequenz ist ein ganzzahliges Vielfaches dieser Sekunde (s. Abb. 1), dann würde die Anlage für das Analysesystem quasi stillstehen oder offensichtlich verklemmt sein.

Die einzige Abhilfe hier wäre eine Erhöhung der Sample-Frequenz (s. Abb. 2). Da diese Maschinen oftmals mit variablen Geschwindigkeiten produzieren können, müsste man diese allerdings auf ein Vielfaches der Maximalgeschwindigkeit auslegen. Wenn die Presse sich nicht gleichmäßig, sondern nur innerhalb eines Teils des Produktionszyklus bewegt, wird es noch schlimmer. Um wirklich allgemein nutzbare Daten zur Verfügung zu haben, muss hier eine gigantische Datenmenge bearbeitet werden.

Die jungen Wilden bei Apache zur Rettung

Wie bereits erwähnt, unterstützen alle mir bekannten Hardware-Lösungen keine Vorverarbeitung. Hier wäre die einzige Option die Einführung eines nachgelagerten Elements. Man könnte die Daten des Adapters etwa mittels eines zwischengeschalteten Rechners abgreifen, die Rohdaten aufbereiten und erst dann zur Speicherung an die Cloud senden. Hier könnte man auch die Sample-Frequenz in die Höhe schrauben, ohne Angst haben zu müssen, dass der dazu nötige Speicherplatz ins Unermessliche wächst. Allerdings erhöht dies auch die Komplexität und den Administrationsaufwand des Systems, und hier wird in der Regel versucht, beides auf ein Minimum zu reduzieren.

Praktisch wäre es, wenn es einfach zu benutzende Frameworks gäbe, um sich dieser Aufgabe zu stellen. Gerade in diesem Bereich hat sich bei der Apache Software Foundation (ASF) in den letzten zwei Jahren einiges getan.

Apache Edgent (incubating)

Apache Edgent (incubating) (1) ist ein Projekt, welches ursprünglich bei IBM entwickelt wurde – anfänglich unter dem Namen Quarks. Als es dann Anfang 2016 bei der Apache Software Foundation(ASF) in den Apache Incubator aufgenommen wurde, wurde es in Apache Edgent umbenannt.

Bei Apache Edgent handelt es sich um ein Programmiermodell für Streaming-Applikationen sowie einer sehr leichtgewichtigen Runtime, um Analysen auf Edge-Devices durchführen zu können, welche oftmals über sehr beschränkte Ressourcen verfügen.

Zwar gibt es momentan Streaming-Frameworks wie Sand am Meer, etwa Apache Storm, Apache Spark oder Apache Flink, allerdings konzentrieren sich diese meist auf das Verarbeiten von großen bis gigantischen Datenmengen durch den Einsatz in skalierbaren Systemen mittels Verteilung der Last auf mehrere Rechner.

Apache Edgent hat hier einen komplett anderen Ansatz, denn es ist nicht als Cluster-System konzipiert, sondern läuft auf einzelnen Systemen, so nah wie möglich an der Datenquelle. Daher muss es sehr schlank, leichtgewichtig und einfach zu konfigurieren sein. Somit können keine gigantischen Berechnungen auf einem einzelnen Cluster-Knoten durchgeführt werden, das ist aber auch nicht der Fokus, denn beim Edge Computing geht es vielmehr um einfachere Filter- und Aggregationsoperationen. Man kann sich das wie einen Türsteher vorstellen, der Einlasskontrollen durchführt und gegebenenfalls eine eingeschränkte Vorverarbeitung von Daten vornimmt, bevor diese an ein größeres System weiter geleitet werden.

Apache Edgent hat daher auch eine Vielzahl von Funktionen, die gerade in diesem Bereich wichtig sind. Vor allem Aggregationund Filterfunktionen sind hier hilfreich, aber auch Fan-out- und Fan-in-Operationen erlauben es, den Datenfluss zu steuern und so komplexe Streaming-Netzwerke aufzubauen. Darüber hinaus sind auch Adapter zum Lesen und Schreiben von Daten via MQTT und Apache Kafka enthalten.

Ideal für Automotive

Vor allem im Automotive-Bereich bietet sich der Einsatz von Apache Edgent an. Deshalb beziehen sich in der Dokumentation auch viele Beispiele auf eben solche Automotive-Szenarien, auch das von mir gewählte Beispiel. Der Grund hierfür ist, dass in modernen Fahrzeugen das zunehmend dominierende Media-Center auch IoT-Aufgaben übernimmt. Diese sind nämlich meistens so dimensioniert, dass darauf auch komplexere Anwendungen ausgeführt werden können. Allerdings wird diese Kapazität nur selten ausgenutzt und daher können diese auch für IoT-Zwecke genutzt werden. Darüber hinaus handelt es sich üblicherweise nicht um für die Kernfunktionen des Fahrzeugs essenzielle Systeme, die trotzdem im normalen Betrieb des Fahrzeugs immer verfügbar sind. Würde die IoT-Anwendung allerdings ständig die zur Verfügung stehenden Kapazitäten auslasten, so wäre davon wahrscheinlich kein Kunde begeistert. Apache Edgent ermöglicht es hier, den Ressourcenaufwand so klein wie möglich zu halten.

Handelt es sich um eine Lösung, welche via mobile Daten mit dem Hersteller oder mit anderen Fahrzeugen kommuniziert (Connected Cars), so entlastet die Möglichkeit zum Filtern, Aggregieren und Zwischenspeichern von Daten diese Datenverbindung sehr.

Praxisbeispiel einer Apache-Edgent-Applikation

Eine Apache-Edgent-Applikation besteht in der Regel aus einem sogenannten „TopologyProvider“. Hierbei handelt es sich im Grunde um die Edgent Runtime. Edgent bringt dabei genau drei Typen von TopologyProvidern mit:

• DirectProvider: Provider für den normalen Einsatz.
• DevelopmentProvider: Spezieller DirectProvider, der zur Entwicklungszeit auch eine Weboberfläche bereitstellt, die den Zustand der damit erzeugten Streams sichtbar macht.
• IoTProvider: Spezieller Provider, der auf die Bedingungen in IoT-Devices optimiert ist, bei dem sich die Edgent-Applikation mit einem sogenannten „IoT-Message Hub“ verbindet. Edgent liefert hier die nötigen Adapter für die Nutzung der „IBM Watson IoT Platform“ gleich mit.

In folgendem Beispiel nutzen wir den DevelopmentProvider sowie simulierte Datenquellen, um eine Edgent-Applikation zu bauen:

1DirectProvider dp = new DevelopmentProvider(); 
2Topology top = dp.newTopology("TemperatureMonitor");

In dem Code-Beispiel wird zuerst ein TopologyProvider erzeugt, mit dessen Hilfe dann eine sogenannte Topologie erzeugt wird. Dieser geben wir den Namen „TemperatureMonitor“. Wie bereits erwähnt, bietet der DevelopmentProvider eine Weboberfläche zur Anzeige des Zustands aller Streams. Mithilfe des folgenden Befehls geben wir die URL dazu aus:

1System.out.println(dp.getServices().getService(
2    HttpServer.class).getConsoleUrl());

Angenommen, wir möchten nun Daten eines Temperatursensors abgreifen. In Ermangelung eines echten Sensors liefert Edgent als Teil seiner Samples auch simulierte Sensoren an:

• SimpleSimulatedSensor: Liefert Werte anhand eines Startwertes, bei dem jeder Folgewert maximal einen bestimmten Wert vom vorherigen abweicht und innerhalb gegebener Wertebereiche bleibt.
• SimulatedGpsSensor: Simuliert GPS-Geokoordinaten einer Fahrtvom IBM Silicon Valey Lab zu IBM Almaden Research und zurück.
• SimulatedTemperatureSensor: Eigentlich nur ein SimpleSimulatedSensor mit einfachen Defaultwerten.

Ein Stream in Edgent besteht aus einer Streaming-Quelle, beliebig vielen Filtern und einer Senke. Wir verwenden in unserem Beispiel einen SimulatedTemperatureSensor als Quelle:

1SimulatedTemperatureSensor tempSensor = 
2    new SimulatedTemperatureSensor(79.0, Ranges.closed(28.0, 112.0), 1.0);

Dieser erzeugt nun einen simulierten Temperatursensor, der mit 79°C beginnend auf Anfrage Werte zwischen 28°C und 112°C zurückgibt, wobei jeder neue Wert maximal 1°C vom vorherigen abweicht. Allerdings erzeugt dieser nicht pro-aktiv Daten, sondern nur auf Anfrage. Nutzt man zum Beispiel einen Kafka-Stream als Quelle, so würde dieser aktiv Events erzeugen. Um dennoch mit einem derartigen, passiven „Sensor“ arbeiten zu können, müssen Daten angefragt werden. Dazu bringt eine Edgent-Topologie entsprechende Helfer mit, die mit folgendem Code initialisiert werden:

1TStream<Double> tempReadings = top.poll(tempSensor, 1, TimeUnit.SECONDS);

Edgent würde nun einmal pro Sekunde einen Wert am Temperatursensor abfragen und daraus einen Datenstrom aus Double-Werten erzeugen.

In der Mess- und Regelungstechnik kommt ein Phänomen nicht selten vor: Messfehler. Mit diesen müssen wir entsprechend umgehen können, und nur, weil eine Messung einmalig oberhalb eines Maximalwertes lag, sollte sich das Fahrzeug nicht automatisch abschalten und einen Abschleppwagen bestellen. Vielmehr sollte hier der gleitende Durchschnitt über eine bestimmte Zeit herangezogen werden. Auch dazu bietet Edgent hilfreiche Funktionen an:

1TWindow<Double, Integer> window = 
2    tempReadings.last( 5, TimeUnit.SECONDS, tuple -> 0); 
3TStream<Double> averageTemperatures = window.aggregate(
4    (tuples, key) -> Precision.round(tuples.stream()
5        .mapToDouble(a -> a).average().getAsDouble(), 1));

Hier wird zuerst aus dem Double Stream ein Stream aus Windows gemacht, die jeweils die Werte der letzten 5 Sekunden enthalten. Danach wird aus diesen Windows wieder ein Double Stream mit den jeweiligen Durchschnittswerten gemacht. Somit wird der Datenstrom geglättet und gelegentliche Messfehler führen nicht mehr zu abrupten Reaktionen, und dennoch würde das Fahrzeug innerhalb von fünf Sekunden etwa eine tatsächlich auftretende Überhitzungen erkennen. Um dieses Verhalten zu implementieren, müssen zuerst die Normalwerte aus diesem Stream herausgefiltert werden:

1TStream<Double> overheated = averageTemperatures.filter( temp -> temp >= 80);

Der „overheated“ Stream sollte nun nur noch in dem Fall einen Event erhalten, wenn die Durchschnittstemperatur der letzten 5 Sekunden bei mindestens 80°C lag. In diesem Fall soll das Fahrzeug nun aber einen Hilferuf absetzen:

1SimulatedGpsSensor gpsSensor = new SimulatedGpsSensor();
2TStream<JsonObject> overheatReports = overheated .map(avgTemp -> { 
3    JsonObject jObj = new JsonObject(); 
4    jObj.addProperty("time", Calendar.getInstance().toString()); 
5    jObj.addProperty("carId", CAR_ID); 
6    jObj.addProperty("temperature", avgTemp);
7    GpsSensor gpsSensorReading = gpsSensor.nextGps(); 
8    JsonObject location = new JsonObject(); 
9    location.addProperty("lon", gpsSensorReading.getLongitude()); 
10    location.addProperty("lat", gpsSensorReading.getLatitude()); 
11    jObj.add("location", location);
12    return jObj;
13});

In der ersten Zeile wird der bereits erwähnte GPS-Sensor initialisiert. Auch dieser liefert nur auf Anfrage neue Werte. Im Rest wird für jeden Wert des „Überhitzte-Durchschnittstemperatur“-Streams ein JsonObject erzeugt, welches alle für den Hilferuf nötigen Informationen enthält. Zu guter Letzt muss diese Information noch ausgegeben werden. In diesem Beispiel geben wir die Daten einfach auf der Konsole aus:

1overheatReports.sink(
2    overheatReport -> System.out.println("JSON: " + overheatReport));

Somit ist unsere erste einfache Edgent-Topologie fertiggestellt. Um sie zu starten, muss sie lediglich noch „submitted” werden:

1dp.submit(top);

Allerdings wird in diesem Fall sehr wenig auf der Konsole ausgegeben. Um zu sehen, was in diesem Fall passiert, können wir in die Topologie noch einige „Sinks“ einbauen, die beispielsweise die gemessenen Werte ausgeben. Dazu würden wir den folgenden Code vor dem obigen Submit der Topologie einfügen:

1tempReadings.sink(curTemp -> System.out.println("Cur Temp: " + curTemp)); 
2averageTemperatures.sink(
3    avgTemp -> System.out.println( "Avg Temp: " + avgTemp));

Das komplette Codebeispiel sieht damit wie folgt aus:

1DirectProvider dp = new DevelopmentProvider(); 
2Topology top = dp.newTopology("TemperatureMonitor");
3System.out.println( dp.getServices().getService(
4    HttpServer.class).getConsoleUrl());
5 
6// Generate a stream of temperature readings. 
7SimulatedTemperatureSensor tempSensor = 
8    new SimulatedTemperatureSensor(79.0, Ranges.closed(28.0, 112.0), 1.0);
9// Poll a temperature values once a second.
10TStream<Double> tempReadings = top.poll(tempSensor, 1, TimeUnit.SECONDS);
11// Generate a stream of values representing the average
12// temperature of the last 5 seconds.
13TWindow<Double, Integer> window = 
14    tempReadings.last( 5, TimeUnit.SECONDS, tuple -> 0); 
15TStream<Double> averageTemperatures = window.aggregate(
16    (tuples, key) -> Precision.round(tuples.stream()
17        .mapToDouble(a -> a) .average().getAsDouble(), 1) );
18// Filter out all the values below 80°C
19TStream<Double> overheated = averageTemperatures.filter( temp -> temp >= 80);
20// Initialize the gps sensor. 
21SimulatedGpsSensor gpsSensor = new SimulatedGpsSensor();
22// Send out a message containing the car id, the temperature and
23// the GPS coordinates where the car currently is located.
24TStream<JsonObject> overheatReports = overheated .map(avgTemp -> { 
25    JsonObject jObj = new JsonObject(); 
26    jObj.addProperty("time", Calendar.getInstance().getTime().toString());
27    jObj.addProperty("carId", CAR_ID); 
28    jObj.addProperty("temperature", avgTemp);
29    GpsSensor gpsSensorReading = gpsSensor.nextGps(); 
30    JsonObject location = new JsonObject(); 
31    location.addProperty("lon", gpsSensorReading.getLongitude()); 
32    location.addProperty("lat", gpsSensorReading.getLatitude()); 
33    jObj.add("location", location);
34    return jObj;
35});
36// Output each overheat report on the console.
37tempReadings.sink(curTemp -> System.out.println("Cur Temp: " + curTemp)); 
38averageTemperatures.sink(
39    avgTemp -> System.out.println("Avg Temp: " + avgTemp)); 
40overheatReports.sink(
41    overheatReport -> System.out.println("JSON: " + overheatReport));
42// Finally submit the just created topology to the runtime. 
43dp.submit(top);

Ich musste die Parameter ein wenig tunen, da ich sonst sehr lange auf Überhitzungen warten musste, mit obigem Code sieht der Output bei mir in etwa folgendermaßen aus:

1Cur Temp: 79.9
2Avg Temp: 79.1
3Cur Temp: 80.5
4Avg Temp: 79.2
5Cur Temp: 79.8
6Avg Temp: 79.5
7Avg Temp: 79.8
8Avg Temp: 79.9
9Cur Temp: 80.2
10JSON: {"time":"Thu Feb 15 14:13:10 CET 2018", "carId":"mycoolcar",
11    "temperature":80.1,"location":{"lon":-121.748087,"lat":37.195647}}
12Avg Temp: 80.1

Ich hoffe, ich konnte mit diesem kleinen Beispiel zeigen, wie einfach es ist, Anwendungen mit Apache Edgent zu entwickeln.

Ein weiterer Vorteil von Apache Edgent ist, dass sich die Topologien und Event-Streams zur Laufzeit starten und stoppen sowie Parameter ändern lassen. Somit ließe sich beispielsweise in obigem Beispiel bei Überhitzung ein Diagnose-Stream aktivieren, der weitere Daten sammelt und im Hilferuf gleich noch eine erste Analyse des Problems mitliefert, sodass beispielsweise der Techniker gleich schon bestimmte Ersatzteile und Werkzeuge mitbringen kann.

Apache Edgent in der Industrie

Bisher habe ich aufgezeigt, wie einfach es ist, mit einer Edgent-Applikation die Vorverarbeitung und Filterung der vom System erzeugten Daten zu übernehmen. Wenn es nun um die Datenerfassung im Bereich der Industrie geht, so gibt es aus meiner Sicht noch ein großes Problem zu lösen: Wie komme ich an die Daten?

In der Industrie wird in der Regel mit Steuerungsgeräten kommuniziert, bei denen der Zugang zu den darin enthaltenen Daten schwierig ist und daher meistens auf zusätzliche Geräte zurückgegriffen werden muss.

Nun möchte ich aus meiner Sicht erklären, warum das momentan noch so problematisch ist: In der Industrie wird üblicherweise auf sogenannte (Speicher-) Programmierbare Steuerungen (SPS) (Engl: Programmable Logic Controler, PLC) zurückgegriffen. Hier gibt es nur ganz vereinzelt Geräte, die offene Protokolle unterstützen, auf welche wir direkt programmatisch zugreifen können. Einige Modelle können etwa mittels des offenen Standards MQTT angesprochen werden. Allerdings sind diese Geräte verhältnismäßig neu und selten, sodass man diese in realen Szenarien noch sehr selten antrifft und eher vom Gegenteil ausgehen sollte.

Um alles noch ein wenig mehr zu erschweren, gibt es zwar Standards, aber leider sehr viele davon, und deren Verbreitung ist, global gesehen, sehr ungleichmäßig. Protokolle, die in Europa gängig sind, sind in den USA eher Nischenprodukte und umgekehrt. Im asiatischen Raum haben sich wieder andere Protokolle etabliert. Bisher war der gebräuchlichste Weg, über einen Protokolladapter auf die Daten in den Steuerungen zuzugreifen. Um wirklich universell nutzbar zu sein, müssen hier aber sehr viele Daten, sehr häufig von den angeschlossenen Steuerungen, abgerufen werden. Auch erhöht dieser zusätzliche Schritt die Latenz, mit der neue Daten zur Verfügung stehen.

Eine andere Alternative wäre es, einen Softwaretreiber für das gewünschte Protokoll zu verwenden. Hier gibt es die Option, einen kommerziellen Treiber einzukaufen oder einen der verfügbaren Open-Source-Treiber zu verwenden. Letztere hatten aber üblicherweise den Nachteil, sehr alt zu sein, unter erheblichen Nebenläufigkeitsproblemen zu leiden und unter der GPL (oder ähnlich) lizenziert zu sein, was einen industriellen Einsatz quasi ausschließt. Außerdem wäre die Software nicht Protokoll übergreifend einsetzbar, denn eine Anpassung an einen anderen Steuerungstyp oder ein anderes Protokoll würde eine Neuimplementierung eines großen Teils der Anwendung bedeuten.

Apache PLC4X (incubating)

Genau hier kommt das zweite Projekt der jungen Wilden von Apache ins Spiel. Ursprünglich von mir als Innovationsprojekt bei der codecentric AG initiiert und seit Dezember 2017 als Apache PLC4X (incubating) Teil des Apache-Incubator-Projekts, hat es sich zum Ziel gesetzt, eine brauchbare Open-Source-Lösung für die Kommunikation mit Programmierbaren Steuerungen (daher das „PLC“ im Namen) im kommerziellen und industriellen Umfeld bereitzustellen. Im Gegensatz zu anderen Alternativen ermöglicht es Apache PLC4X, Programme für den Zugriff auf Steuerungen zu implementieren, ohne sich dabei auf eine Steuerung oder ein verwendetes Protokoll festlegen zu müssen. Alle Treiber werden über die gleiche Programmierschnittstelle angesprochen, sind aber in verschiedenen Sprachen wie Java, Scala oder später auch mal C und C++ verfügbar (daher das „4X“ im Namen). Die Umstellung auf eine andere Steuerung oder ein anderes Protokoll erfordert hier lediglich den Austausch des Treibers sowie die eventuelle Anpassung der Connection- und Adressstrings.

Damit lässt sich Apache PLC4X wunderbar in Apache-Edgent-Applikationen einsetzen, um auf die Daten der Steuerungen zuzugreifen. In diesem Fall kann auf den Einsatz von Protokolladaptern komplett verzichtet werden. Genau aus diesem Grund bietet PLC4X auch gleich eine entsprechende Apache-Edgent-Integration mit an. Folgendes Listing demonstriert, wie einfach sich Apache PLC4X in einem Apache-Edgent-Programm einsetzen lässt:

1// Baue eine Verbindung zu der S7-Steuerung auf. 
2PlcConnectionAdapter plcAdapter = 
3    new PlcConnectionAdapter("s7://192.168.0.1/1/1");
4// Initialisiere das Edgent System. 
5DirectProvider dp = new DirectProvider(); 
6Topology top = dp.newTopology("kafka-bridge");
7// Erzeuge einen Edgent Supplier welcher einen Stream aus 
8// Bytes erzeugt. Die übergebene Adresse liest den Zustand 
9// der ersten 8 Digital-Eingänge einer handelsüblichen 
10// Siemens S7 1200 Steuerung aus. 
11Supplier<Byte> digitalInputSupplier = 
12    PlcFunctions.byteSupplier(plcAdapter, "INPUTS/0");
13// Polle den eben erzeugten Supplier ein Mal pro Sekunde. 
14TStream<Byte> source = top.poll(digitalInputSupplier, 1, TimeUnit.SECONDS);
15// Da Kafka keine Bytes zu mögen scheint: 
16// Konvertiere die Bytes in Strings. 
17TStream<String> stringSource = source.map(value -> Byte.toString(value));
18// Erzeuge einen Kafka Producer (Der Kafka Nachrichten 
19// aus Events in Edgent Streams erzeugt) 
20Map<String,Object> kafkaConfig = new HashMap<>(); 
21kafkaConfig.put("bootstrap.servers", "localhost:9092"); 
22KafkaProducer kafka = new KafkaProducer(top, () -> kafkaConfig);
23// Veröffentliche den eben erzeugten String Stream in Kafka. 
24kafka.publish(stringSource, "digital-inputs"); 
25// Starte die Topologie. 
26dp.submit(top);

Vom Grundaufbau her gleicht die Anwendung unserem ersten Beispiel. Hier verwende ich allerdings statt des DevelpmentProviders den für Produktiveinsätze besser geeigneten DirectProvider.

Ganz am Anfang wird eine Verbindung zu einer Siemens-S7-Steuerung mit der IP-Adresse 192.168.0.1 aufgebaut. Diese Verbindung wird nun für alle weiteren Operationen verwendet. Der nächste Unterschied ist, dass statt des SimulatedTemperatureSensors mittels der PlcFunctions-Klasse eine echte Datenquelle erzeugt wird, welche von der Steuerung jeweils ein Byte von der Adresse „INPUTS/0“ liest. Hierbei handelt es sich um die Zustände der ersten acht Digitaleingänge der Steuerung. Jedes Mal, wenn von dieser Datenquelle gelesen wird, wird also der gewünschte Wert aus der Steuerung gelesen und an den Datenstrom geschickt.

Auch hier nutzen wir zurzeit noch die Polling-Funktion von Edgent. Wir planen, dass in naher Zukunft die Treiber auch Publish-/ Subscribe-Features anbieten. In diesem Fall würde automatisch ein Byte an den Stream geschickt, sowie Daten von der Steuerung eintreffen. Damit würde das Pollen entfallen können. Im Datenstrom werden dann die Byte-Werte in Strings konvertiert, bevor diese dann an einen „KafkaProducer“ geschickt werden.

Das mag seltsam klingen, allerdings ist dieser Edgent-Consumer in der Tat ein Kafka-Producer, da er Kafka-Nachrichten produziert. Dieses Beispiel stellt also quasi einen S7-Kafka-Protokolladapter dar, der zu 100 Prozent aus Open Source besteht.

Fazit

Mit den beiden Incubator-Projekten der Apache Software Foundation Apache Edgent und Apache PLC4X ist es nun möglich, auf einem Edge-Device genau die Daten abzurufen, die benötigt werden, und das auch genau, wann sie benötigt werden. Durch die lokale Vorverarbeitung der Daten werden nur noch die nützlichen Informationen in einer aufbereiteten Form an die nachgelagerten Systeme weitergeleitet. Dies entlastet nicht nur die Programmierbaren Steuerungen, sondern auch die Netze und vor allem die Systeme, die die Daten entgegennehmen, speichern und verarbeiten.

1) Es sei auch erwähnt, was es mit dem „incubating“ auf sich hat: Bei Apache Edgent und Apache PLC4X handelt es sich um Projekte, welche sich bei der Apache Software Foundation (ASF) im „Inkubations-Zustand“, unter der Aufsicht des Apache Incubator PMCs (Project Management Comittee) befinden. Dies wird von der ASF von allen neuen Projekten verlangt, bis diese unter Beweis gestellt haben, dass sich ihre Infrastruktur, Kommunikationskultur und Entscheidungsfindungsprozesse auf eine Art stabilisiert haben, die konsistent ist mit denen anderer erfolgreicher ASF-Projekte. Der Inkubations-Status ist folglich nicht zwangsläufig ein Indikator für Vollständigkeit und Stabilität des Codes, sondern besagt vielmehr, dass sich das Projekt als Ganzes noch die volle Unterstützung der ASF verdienen muss.