{"id":1295,"date":"2023-05-13T08:09:48","date_gmt":"2023-05-13T06:09:48","guid":{"rendered":"https:\/\/www.rocworks.at\/wordpress\/?p=1295"},"modified":"2023-05-13T09:07:33","modified_gmt":"2023-05-13T07:07:33","slug":"scada-real-time-data-and-apache-spark","status":"publish","type":"post","link":"https:\/\/www.rocworks.at\/wordpress\/?p=1295","title":{"rendered":"SCADA Real Time Data and Apache SPARK"},"content":{"rendered":"\n

The integration of SCADA with Spark<\/strong> and WinCC Open Architecture offers a powerful and versatile solution that combines real-time data<\/strong> processing, advanced analytics<\/strong>, scalability<\/strong>, and flexibility. This combination empowers you to optimize industrial processes<\/strong>, make data-driven decisions<\/strong>, and stay ahead in a rapidly evolving technological landscape.<\/p>\n\n\n\n

By utilizing my 5-year-old project that implemented a native Java manager for WinCC Open Architecture<\/a>, I have enabled the integration of SCADA with Spark <\/strong>for the current WinCC OA Version 3.19. <\/p>\n\n\n\n

Very simple example is to analyze tags and the corresponding amount of values in your SCADA system can provide valuable insights into the distribution and characteristics of the data.<\/p>\n\n\n\n

res = spark.sql('SELECT tag, count(*) count FROM events GROUP BY tag ORDER by count(*) DESC')\ndata = res.toPandas()\nplt.figure( figsize = ( 10, 6 ) )\nsns.barplot( x=\"count\", y=\"tag\", data=data)\nplt.show()<\/code><\/pre>\n\n\n\n
\"\"<\/a><\/figure>\n\n\n\n
Another simple example is to calculate the moving average of 10 preceding and following values for a given data point in a time series, you can use a sliding window approach:<\/p>\n\n\n\n
data = spark.sql(\"\"\"\nSELECT ROUND(value,2) as value, \n  AVG(value) OVER (PARTITION BY tag ORDER BY ts \n  ROWS BETWEEN 10 PRECEDING AND 10 FOLLOWING) avg \n  FROM events \n  WHERE tag = 'System1:ExampleDP_Trend2.'\n  ORDER BY ts DESC\n  LIMIT 100\n  \"\"\").toPandas()\ndata = data.reset_index().rename(columns={\"index\": \"nr\"})\nsns.lineplot(data=data, x='nr', y='value', label='Value')\nsns.lineplot(data=data, x='nr', y='avg', label='Average')\nplt.show()<\/code><\/pre>\n\n\n\n
\"\"<\/a><\/figure>\n\n\n\n
By leveraging the distributed file system<\/strong>, you can take advantage of Spark’s parallel processing capabilities. The distributed file system ensures that the data frame is partitioned<\/strong> and distributed<\/strong> across the nodes<\/strong> of the Spark cluster<\/strong>, enabling simultaneous processing<\/strong> of data in parallel<\/strong>. This distributed approach enhances performance<\/strong> and scalability<\/strong>, allowing for efficient handling of large volumes of real-time SCADA data<\/strong>.<\/p>\n\n\n\n
I have achieved real-time data streaming<\/strong> from WinCC OA to a Spark cluster with a Websocket-Server<\/a> based on the Java manager. This streaming process involves continuously<\/strong> transferring<\/strong> SCADA real-time data<\/strong> from the WinCC OA system to the Spark <\/strong>cluster for further processing and analysis.<\/p>\n\n\n\n
url='wss:\/\/192.168.1.190:8443\/winccoa?username=root&password='\nws = create_connection(url, sslopt={\"cert_reqs\": ssl.CERT_NONE})\n\ndef read():\n    while True:\n        on_message(ws.recv())\nThread(target=read).start()\n\ncmd={'DpQueryConnect': {'Id': 1, 'Query':\"SELECT '_online.._value' FROM 'ExampleDP_*.'\", 'Answer': False}}\nws.send(json.dumps(cmd))<\/code><\/pre>\n\n\n\n
Once the data is received by the Spark cluster, I store it as a data frame on the distributed file system (DFS). A data frame is a distributed collection of data organized into named columns, similar to a table in a relational database. Storing the data frame on the distributed file system ensures data persistence and allows for efficient processing and retrieval.<\/p>\n\n\n\n
schema = StructType([\n    StructField(\"ts\", TimestampType(), nullable=False),\n    StructField(\"tag\", StringType(), nullable=False),\n    StructField(\"value\", FloatType(), nullable=False)\n])\ndf = spark.createDataFrame(spark.sparkContext.emptyRDD(), schema)<\/code><\/pre>\n\n\n\n
bulk = []\nlast = datetime.datetime.now()\ndef on_message(message):\n    global bulk, last, start\n    data = json.loads(message) \n            \n    if \"DpQueryConnectResult\" in data:\n        values = data[\"DpQueryConnectResult\"][\"Values\"]\n        for tag, value in values:\n            #print(tag, value)\n            data = {\"ts\": datetime.datetime.now(), \"tag\": tag, \"value\": value}\n            bulk.append(data)\n            \n    now =datetime.datetime.now()\n    time = datetime.datetime.now() - last\n    if time.total_seconds() > 10 or len(bulk) >= 1000:\n        last = now\n\n        # Create a new DataFrame with the received data\n        new_df = spark.createDataFrame(bulk, schema)\n        \n        new_df.write \\\n            .format(\"csv\") \\\n            .option(\"header\", \"true\") \\\n            .mode(\"append\") \\\n            .save(\"events.csv\")\n        \n        bulk = []<\/code><\/pre>\n\n\n\n
Once the SCADA data is stored as a distributed data frame on the Spark cluster’s distributed file system, you can leverage Spark’s parallel processing capabilities to efficiently process the data in parallel.<\/p>\n\n\n\n
df = spark.read \\\n    .format(\"csv\") \\\n    .option(\"header\", \"true\") \\\n    .option(\"timezone\", \"UTC\") \\\n    .schema(schema) \\\n    .load(\"events.csv\")\ndf.createOrReplaceTempView(\"events\")<\/code><\/pre>\n\n\n\n
By combining SCADA (Supervisory Control and Data Acquisition) with Spark’s powerful data processing capabilities, I have created a solution that can handle large volumes of real-time data efficiently. This enables faster and more accurate decision-making based on the insights derived from the processed data.<\/p>\n","protected":false},"excerpt":{"rendered":"
The integration of SCADA with Spark and WinCC Open Architecture offers a powerful and versatile solution that combines real-time data processing, advanced analytics, scalability, and flexibility. This combination empowers you to optimize industrial processes, make data-driven decisions, and stay ahead in a rapidly evolving technological landscape. By utilizing my 5-year-old project that implemented a native … Continue reading SCADA Real Time Data and Apache SPARK<\/span> →<\/span><\/a><\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"closed","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1,29,3],"tags":[],"class_list":["post-1295","post","type-post","status-publish","format-standard","hentry","category-allgemein","category-bigdatanosql","category-wincc-oa"],"_links":{"self":[{"href":"https:\/\/www.rocworks.at\/wordpress\/index.php?rest_route=\/wp\/v2\/posts\/1295","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/www.rocworks.at\/wordpress\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.rocworks.at\/wordpress\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.rocworks.at\/wordpress\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/www.rocworks.at\/wordpress\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=1295"}],"version-history":[{"count":7,"href":"https:\/\/www.rocworks.at\/wordpress\/index.php?rest_route=\/wp\/v2\/posts\/1295\/revisions"}],"predecessor-version":[{"id":1305,"href":"https:\/\/www.rocworks.at\/wordpress\/index.php?rest_route=\/wp\/v2\/posts\/1295\/revisions\/1305"}],"wp:attachment":[{"href":"https:\/\/www.rocworks.at\/wordpress\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=1295"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.rocworks.at\/wordpress\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=1295"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.rocworks.at\/wordpress\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=1295"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}