\n\n
In [ ]:<\/div>\n
<\/span>!<\/span>pip install pyspark\n<\/pre><\/div>\n\n    <\/div>\n<\/div>\n<\/div>\n\n<\/div>\n\n\nIn [ ]:<\/div>\n\n    \n<\/span>from<\/span> pyspark<\/span> import<\/span> SparkContext<\/span>,<\/span> SparkConf<\/span>\nfrom<\/span> pyspark.sql<\/span> import<\/span> SparkSession<\/span>,<\/span> SQLContext<\/span>\n\nsc<\/span> =<\/span> SparkContext<\/span>(<\/span>'local'<\/span>,<\/span> 'ml-studio'<\/span>)<\/span>\nsc<\/span>\n<\/pre><\/div>\n\n    <\/div>\n<\/div>\n<\/div>\n\n<\/div>\n\n\nIn [ ]:<\/div>\n\n    \n<\/span>sqlContext<\/span> =<\/span> SQLContext<\/span>(<\/span>sc<\/span>)<\/span>\nsqlContext<\/span>\n<\/pre><\/div>\n\n    <\/div>\n<\/div>\n<\/div>\n\n<\/div>\n\n<\/div>\n\nOutlier Detection: An ETL Tutorial with Spark¶<\/a><\/h1>Part of the Industry 4.0 framework is to make sure that manufacturers have more visibility over what\u2019s going on with their machines in the factory floors. This is why Industry 4.0 works tightly with Internet of Things. IoT allows large scale real-time data collection from sensors that are installed in production equipment possible. Nevertheless, having good Data Collection Agents alone isn\u2019t sufficient. We need an automated way of extracting, analyzing and summarizing information from the large data stream, since it\u2019s impossible for humans to do it manually. In big data terminology, this process is often referred to as ETL (Extract-Transform-Load).<\/p>\n
Today, we\u2019ll discuss one family of algorithm that I have personally seen to be useful in the industry: outlier detection. The idea is to find any abnormal measurements from the data stream and highlight them to the domain experts e.g. process engineers. I will share an implementation of a basic anomaly detection algorithm in Spark. I could have done the same tutorial with python\u2019s pandas-dataframe, but unfortunately once we deal with big dataset (whose size is way larger than memory space), the latter is no longer suitable.<\/p>\n
This is a simple dummy dataset that I use. Suppose we have data stream from 2 sensors. How can we automatically capture the two anomalous dots that are present below?<\/p>\n<\/center>Outlier Detection¶<\/a><\/h2>The model that we use finds region of values whose probability of occurrence are low under the distribution that has been fitted to the observed data. We assume that our sensors are unimodal gaussian in nature. With that, we can calculate the two thresholds that are six sigma away from the distribution\u2019s mean.<\/p>\n<\/center>Visually, the thresholds are fitted in this manner. Any measurements above the Upper Limit (around 25) or below the Lower Limit (around 15) are deemed as outliers.<\/p>\n<\/center>Now to implement this in Spark, we first import all of the library dependencies:<\/p>\n\n<\/div>\n<\/div>\n<\/div>\n\n\nIn [ ]:<\/div>\n\n    \n<\/span>from<\/span> pyspark.sql<\/span> import<\/span> *<\/span>\nfrom<\/span> pyspark.sql.types<\/span> import<\/span> *<\/span>\nfrom<\/span> pyspark.sql.functions<\/span> import<\/span> *<\/span>\n<\/pre><\/div>\n\n    <\/div>\n<\/div>\n<\/div>\n\n<\/div>\n\n<\/div>\n\nExtract¶<\/a><\/h2>We now assume that our data comes in a csv format. It has also been saved in a file called test.csv. We first specify the data schema explicitly. Note than in production, data could also be obtained from a database and message broker (e.g. MQTT, Kafka etc\u2026).<\/p>\n\n<\/div>\n<\/div>\n<\/div>\n\n\nIn [ ]:<\/div>\n\n    \n<\/span>customSchema<\/span> =<\/span> StructType<\/span>([<\/span>\n    StructField<\/span>(<\/span>"sensorId"<\/span>,<\/span> StringType<\/span>(),<\/span> True<\/span>),<\/span>\n    StructField<\/span>(<\/span>"values"<\/span>,<\/span> DoubleType<\/span>(),<\/span> True<\/span>)])<\/span>\n<\/pre><\/div>\n\n    <\/div>\n<\/div>\n<\/div>\n\n<\/div>\n\n<\/div>\n\nThen, we read the csv file into a Spark DataFrame. Here we can see that there are only two columns: sensorId and values.<\/p>\n\n<\/div>\n<\/div>\n<\/div>\n
\n\nIn [ ]:<\/div>\n\n    \n<\/span># load dataset<\/span>\ndf<\/span> =<\/span> sqlContext<\/span>.<\/span>read<\/span>.<\/span>format<\/span>(<\/span>"csv"<\/span>)<\/span>.<\/span>option<\/span>(<\/span>"header"<\/span>,<\/span> "true"<\/span>)<\/span>.<\/span>schema<\/span>(<\/span>customSchema<\/span>)<\/span>.<\/span>load<\/span>(<\/span>"data\/test.csv"<\/span>)<\/span>\n  \ndf<\/span>.<\/span>printSchema<\/span>()<\/span>\n<\/pre><\/div>\n\n    <\/div>\n<\/div>\n<\/div>\n\n<\/div>\n\n<\/div>\n\nWe can register the dataset as a table for SQL-style queries:<\/p>\n\n<\/div>\n<\/div>\n<\/div>\n
\n\nIn [ ]:<\/div>\n\n    \n<\/span>df<\/span>.<\/span>createOrReplaceTempView<\/span>(<\/span>"sensors"<\/span>)<\/span>\n<\/pre><\/div>\n\n    <\/div>\n<\/div>\n<\/div>\n\n<\/div>\n\n\nIn [ ]:<\/div>\n\n    \n<\/span>sqlContext<\/span>.<\/span>sql<\/span>(<\/span>"SELECT * FROM sensors"<\/span>)<\/span>.<\/span>show<\/span>()<\/span>\n<\/pre><\/div>\n\n    <\/div>\n<\/div>\n<\/div>\n\n<\/div>\n\n<\/div>\n\nTransform¶<\/a><\/h2>We would like to calculate the distribution profile for each sensorID, particularly the Upper and Lower Outlier Thresholds. To do that, we need to group the dataframe by sensorId, followed by aggregating each sensor data\u2019s mean and standard deviation accordingly. We can then create 2 new columns, one for each outlier threshold.<\/p>\n\n<\/div>\n<\/div>\n<\/div>\n\n\nIn [ ]:<\/div>\n\n    \n<\/span># calculate statistics<\/span>\nstatsDF<\/span> =<\/span> df<\/span>.<\/span>groupBy<\/span>(<\/span>"sensorId"<\/span>)<\/span>.<\/span>agg<\/span>(<\/span>mean<\/span>(<\/span>df<\/span>.<\/span>values<\/span>)<\/span>.<\/span>alias<\/span>(<\/span>"mean"<\/span>),<\/span> stddev<\/span>(<\/span>df<\/span>.<\/span>values<\/span>)<\/span>.<\/span>alias<\/span>(<\/span>"stddev"<\/span>))<\/span>\n\n# add columns with upper and lower limits<\/span>\nstatsDF<\/span> =<\/span> statsDF<\/span>.<\/span>withColumn<\/span>(<\/span>"UpperLimit"<\/span>,<\/span> statsDF<\/span>.<\/span>mean<\/span> +<\/span> statsDF<\/span>.<\/span>stddev<\/span> *<\/span> 3<\/span>)<\/span>.<\/span>withColumn<\/span>(<\/span>"LowerLimit"<\/span>,<\/span> statsDF<\/span>.<\/span>mean<\/span> -<\/span> statsDF<\/span>.<\/span>stddev<\/span> *<\/span> 3<\/span>)<\/span>\n  \nstatsDF<\/span>.<\/span>printSchema<\/span>()<\/span>\n<\/pre><\/div>\n\n    <\/div>\n<\/div>\n<\/div>\n\n<\/div>\n\n<\/div>\n\nWe would now like to find which sensor readings are anomalous in the original dataframe. Since the information live in two different dataframes, we need to join them using the sensorId column as a common index.<\/p>\n\n<\/div>\n<\/div>\n<\/div>\n
\n\nIn [ ]:<\/div>\n\n    \n<\/span># join the two dataframe<\/span>\njoinDF<\/span> =<\/span> df<\/span>.<\/span>join<\/span>(<\/span>statsDF<\/span>,<\/span> df<\/span>.<\/span>sensorId<\/span> ==<\/span> statsDF<\/span>.<\/span>sensorId<\/span>)<\/span>\n<\/pre><\/div>\n\n    <\/div>\n<\/div>\n<\/div>\n\n<\/div>\n\n<\/div>\n\nLastly, we can filter rows whose values lie beyond the range enclosed by the outlier thresholds. Voila! We managed to capture the two anomalous points.<\/p>\n\n<\/div>\n<\/div>\n<\/div>\n
\n\nIn [ ]:<\/div>\n\n    \n<\/span># outlierDetection<\/span>\ndef<\/span> detect_outlier<\/span>(<\/span>values<\/span>,<\/span> UpperLimit<\/span>,<\/span> LowerLimit<\/span>):<\/span>\n    # outliers are points lying below LowerLimit or above upperLimit<\/span>\n    return<\/span> (<\/span>values<\/span> <<\/span> LowerLimit<\/span>)<\/span> or<\/span> (<\/span>values<\/span> ><\/span> UpperLimit<\/span>)<\/span>\n\nudf_detect_outlier<\/span> =<\/span> udf<\/span>(<\/span>lambda<\/span> values<\/span>,<\/span> UpperLimit<\/span>,<\/span> LowerLimit<\/span>:<\/span> detect_outlier<\/span>(<\/span>values<\/span>,<\/span> UpperLimit<\/span>,<\/span> LowerLimit<\/span>),<\/span> BooleanType<\/span>())<\/span>\n\noutlierDF<\/span> =<\/span> joinDF<\/span>.<\/span>withColumn<\/span>(<\/span>"isOutlier"<\/span>,<\/span> udf_detect_outlier<\/span>(<\/span>joinDF<\/span>.<\/span>values<\/span>,<\/span> joinDF<\/span>.<\/span>UpperLimit<\/span>,<\/span> joinDF<\/span>.<\/span>LowerLimit<\/span>))<\/span>.<\/span>filter<\/span>(<\/span>"isOutlier"<\/span>)<\/span>\n  \noutlierDF<\/span>.<\/span>createOrReplaceTempView<\/span>(<\/span>"outliers"<\/span>)<\/span>\n<\/pre><\/div>\n\n    <\/div>\n<\/div>\n<\/div>\n\n<\/div>\n\n\nIn [ ]:<\/div>\n\n    \n<\/span>sqlContext<\/span>.<\/span>sql<\/span>(<\/span>"SELECT * FROM outliers"<\/span>)<\/span>.<\/span>show<\/span>()<\/span>\n<\/pre><\/div>\n\n    <\/div>\n<\/div>\n<\/div>\n\n<\/div>\n\n<\/div>\n\nConclusion¶<\/a><\/h2>We have seen how a typical ETL pipeline with Spark works, using anomaly detection as the main transformation process. Note that some of the procedures used here is not suitable for production. For example, CSV input and output are not encouraged. Normally we would use Hadoop Distributed File System (HDFS) instead. The latter could be wrapped under a database too e.g. HBase. Nonetheless, the main programming paradigm stays the same.<\/p>\n\n<\/div>\n<\/div>\n<\/div>\n
\n\nIn [ ]:<\/div>\n\n    \n<\/span> \n<\/pre><\/div>\n\n    <\/div>\n<\/div>\n<\/div>\n\n<\/div>\n    <\/div>\n  <\/div>\n","protected":false},"excerpt":{"rendered":"In [ ]: !pip install pyspark In [ ]: from pyspark import SparkContext, SparkConf from pyspark.sql import SparkSession, SQLContext sc = SparkContext(‘local’, ‘ml-studio’) sc In [ ]: sqlContext = SQLContext(sc) sqlContext Outlier Detection: An ETL Tutorial with Spark¶ Part of the Industry 4.0 framework is to make sure that manufacturers have more visibility over what’s going on with their machines in […]<\/p>\n","protected":false},"author":11,"featured_media":0,"parent":0,"menu_order":0,"comment_status":"closed","ping_status":"closed","template":"","meta":{"footnotes":""},"class_list":["post-122","page","type-page","status-publish","hentry"],"_links":{"self":[{"href":"http:\/\/vargas-solar.com\/data-ml-studios\/wp-json\/wp\/v2\/pages\/122","targetHints":{"allow":["GET"]}}],"collection":[{"href":"http:\/\/vargas-solar.com\/data-ml-studios\/wp-json\/wp\/v2\/pages"}],"about":[{"href":"http:\/\/vargas-solar.com\/data-ml-studios\/wp-json\/wp\/v2\/types\/page"}],"author":[{"embeddable":true,"href":"http:\/\/vargas-solar.com\/data-ml-studios\/wp-json\/wp\/v2\/users\/11"}],"replies":[{"embeddable":true,"href":"http:\/\/vargas-solar.com\/data-ml-studios\/wp-json\/wp\/v2\/comments?post=122"}],"version-history":[{"count":3,"href":"http:\/\/vargas-solar.com\/data-ml-studios\/wp-json\/wp\/v2\/pages\/122\/revisions"}],"predecessor-version":[{"id":134,"href":"http:\/\/vargas-solar.com\/data-ml-studios\/wp-json\/wp\/v2\/pages\/122\/revisions\/134"}],"wp:attachment":[{"href":"http:\/\/vargas-solar.com\/data-ml-studios\/wp-json\/wp\/v2\/media?parent=122"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}