読者です 読者をやめる 読者になる 読者になる

Technology Topics by Brains

ブレインズテクノロジーの研究開発機関「未来工場」で働くエンジニアが、先端オープン技術、機械学習×データ分析(異常検知、予兆検知)に関する取組みをご紹介します。

Apache Zeppelin & Spark SQLでサーバのログデータを整形・可視化する

こんにちは。春休みにブレインズテクノロジーのインターンシップに参加した、現在学部4年生の松井です。
インターン中にやったこと、ハマったことなどをまとめてみました。

やったこと

Apache Zeppelinというブラウザ上で動くインタラクティブシェルを用いて、S3に置かれているサーバ4台のメトリクス1週間分のログデータを可視化しました。
可視化までのステップを大雑把に区切ると以下のような感じです。

  1. S3に置かれているログデータを必要な分ローカルに保存
  2. ローカルに保存したデータをDataframeとして読み出し
  3. DataFrameに対し、意味のある情報の抜き出し・データ整形といった処理を行う
  4. SQL文を記述してグラフ出力
Spark SQLについて

Apache Zeppelinには分散処理のフレームワークであるApache Sparkのインタプリタが組み込まれており、データの整形・可視化にはApache Sparkの1コンポーネントであるSpark SQLを使いました。
Spark SQLは、関係データベースと似ているDataFrameというデータ構造を使っています。
DataFrameのAPIScala, Java, Python, Rで利用することができ、今回のコードはすべてScalaで記述しています。

実行環境について

メモリ16GBのMacBook Proで実行しました。
Zeppelinのデフォルトの設定だと、メモリ領域やヒープ領域が足りずOutOfMemoryErrorやStackOverflowErrorが出たので、InterpreterからSparkインタプリタの設定を変更してみましたがどうも上手くいかず。。。
最終的には、設定ファイルconf/zeppelin-env.shに以下の行を追加することで解決しました。

export ZEPPELIN_JAVA_OPTS="-Dspark.executor.memory=8g -Dspark.cores.max=16" # Additional jvm options. for example, export ZEPPELIN_JAVA_OPTS="-Dspark.executor.memory=8g -Dspark.cores.max=16"
export ZEPPELIN_MEM="-Xms4096m -Xmx4096m -Xss4096k -XX:PermSize=256m -XX:MaxPermSize=256m" # Zeppelin jvm mem options Default -Xms1024m -Xmx1024m -XX:MaxPermSize=512m

ちなみに、同ファイルにS3へのアクセスキーなども記述する必要があります。

export AWS_SECRET_ACCESS_KEY=####
export AWS_ACCESS_KEY_ID=####
export AWS_DEFAULT_REGION=####

S3に置かれているログデータをローカルに保存

まず、取得する期間やパスなどを初期変数として定めておきます。この変数は今後も使います。

///初期変数
//startDateのhh/mm/ssは無視されて、各日のデータはすべて取得される
//サーバからローカルに保存する場合も、ローカルからDataFrameに保存する場合も、同じこれらの初期変数を使う
val dayRange = z.input("ログ取得日数", 7).toString.toInt
val baseKey = z.input("S3パス", "Your Server's Path").toString
val baseLocalPath = z.input("保存先パス", "Your Local Path").toString
val startDate = z.input("ログ取得開始日時(yyyy/MM/dd HH:mm:ss)", "2017/02/21 23:00:00").toString

日付のパースなどを行い、4台あるサーバのデータをそれぞれローカルに保存していきます。

///S3からログデータ取得、ローカルに保存
import org.apache.spark.rdd.RDD
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat
import org.apache.hadoop.io.{LongWritable, Text}
import java.text.SimpleDateFormat
import java.util.Calendar
import java.util.TimeZone
import java.util.ArrayList

import org.apache.http.NameValuePair
import org.apache.http.client.entity.UrlEncodedFormEntity
import org.apache.http.client.methods.HttpPost
import org.apache.http.impl.client.DefaultHttpClient
import org.apache.http.message.BasicNameValuePair

//Dateのパース
var sdf:SimpleDateFormat = new SimpleDateFormat("yyyy/MM/dd/HH/mm")
sdf.setTimeZone(TimeZone.getTimeZone("JST"))
var nowdt:Calendar = Calendar.getInstance
nowdt.setTimeZone(TimeZone.getTimeZone("JST"))
if(startDate.length > 0) {
  var startDateTmp = startDate.split(" ")
  var sYear = startDateTmp(0).split("/")(0).toInt
  var sMonth = startDateTmp(0).split("/")(1).toInt-1
  var sDay = startDateTmp(0).split("/")(2).toInt
  var sHour = startDateTmp(1).split(":")(0).toInt
  var sMinute = startDateTmp(1).split(":")(1).toInt
  var sSecond = startDateTmp(1).split(":")(2).toInt
  nowdt.set(sYear,sMonth,sDay,sHour,sMinute,sSecond)
}
var nowdate = sdf.format(nowdt.getTime)

def saveAsTextFile(rdd:RDD[String], localPath:String) = {
  rdd.isEmpty() // InvalidInputException (matches 0 files) 対策
  rdd.saveAsTextFile(localPath,classOf[org.apache.hadoop.io.compress.GzipCodec])
  println("save as text file =>"+localPath)

}

def get(inputPath:String, host:String) = {
  try {
    var textFileRDD = sc.textFile(baseKey+inputPath+"*")
    saveAsTextFile(textFileRDD.repartition(1), baseLocalPath+inputPath+"0")
  } catch {
    case faee:org.apache.hadoop.mapred.FileAlreadyExistsException => throw faee
  }
}

def getDate(year:Int, month:Int, day:Int, add:Int) : String = {
  val resDate = "%tY/%<tm/%<td" format new java.util.GregorianCalendar(year, month-1, day-add);
  return resDate;
}

def zeroPadding2(str:String) : String = {
  var res = str
  if(str.length == 1)
    res = "0"+str
  return res
}

val nowdatesplit = nowdate.split("/")
val nowYear = nowdatesplit(0).toInt
val nowMonth = nowdatesplit(1).toInt
val nowDay = nowdatesplit(2).toInt
val nowHour = nowdatesplit(3).toInt
val nowMinuteStr = nowdatesplit(4).substring(0,1)+"0" // example 27 -> 20

//ログデータを10分ごとに切り出しまとめて保存
try {

  for(addDay <- 0 to (dayRange-1)) {

    for(addHour <- ((24-1) to 0 by -1)) { //(nowHour to 0 by -1)

      for(addMinute <- List(50,40,30,20,10,0)) {

        var analysisDate = getDate(nowYear, nowMonth, nowDay, addDay)
        var fileDate = analysisDate.replaceAll("/","")
        var dateSplit = analysisDate.split("/")
        var year = dateSplit(0)
        var month = dateSplit(1)
        var day = dateSplit(2)

        var hourStr = zeroPadding2(addHour.toString)
        var minuteStr = zeroPadding2(addMinute.toString)
        var minutePath = minuteStr.substring(0,1)

        get("SERVER-MG/"+year+"/"+month+"/"+day+"/metricbeat_"+fileDate+hourStr+minutePath, "SERVER-MG")
        get("SERVER-SR01/"+year+"/"+month+"/"+day+"/metricbeat_"+fileDate+hourStr+minutePath, "SERVER-SR01")
        get("SERVER-SR02/"+year+"/"+month+"/"+day+"/metricbeat_"+fileDate+hourStr+minutePath, "SERVER-SR02")
        get("SERVER-SR03/"+year+"/"+month+"/"+day+"/metricbeat_"+fileDate+hourStr+minutePath, "SERVER-SR03")

      }
    }
  }
} catch {
  // 取得するログの日付を遡り取得済みであれば処理終了
  case faee:org.apache.hadoop.mapred.FileAlreadyExistsException => println("exit : "+faee)
}

ローカルに保存したデータをDataframeとして読み出し

前の処理でローカルに保存したログデータを、まずは1つのDataframeに格納します。

///ローカルに保存してあるログデータをDataFrame形式で読み込み
import org.apache.spark.rdd.RDD
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat
import org.apache.hadoop.io.{LongWritable, Text}
import java.text.SimpleDateFormat
import java.util.Calendar
import java.util.TimeZone
import java.util.ArrayList

import org.apache.http.NameValuePair
import org.apache.http.client.entity.UrlEncodedFormEntity
import org.apache.http.client.methods.HttpPost
import org.apache.http.impl.client.DefaultHttpClient
import org.apache.http.message.BasicNameValuePair

//Dateのパースは同様
var sdf:SimpleDateFormat = new SimpleDateFormat("yyyy/MM/dd/HH/mm")
sdf.setTimeZone(TimeZone.getTimeZone("JST"))
var nowdt:Calendar = Calendar.getInstance
nowdt.setTimeZone(TimeZone.getTimeZone("JST"))
if(startDate.length > 0) {
  var startDateTmp = startDate.split(" ")
  var sYear = startDateTmp(0).split("/")(0).toInt
  var sMonth = startDateTmp(0).split("/")(1).toInt-1
  var sDay = startDateTmp(0).split("/")(2).toInt
  var sHour = startDateTmp(1).split(":")(0).toInt
  var sMinute = startDateTmp(1).split(":")(1).toInt
  var sSecond = startDateTmp(1).split(":")(2).toInt
  nowdt.set(sYear,sMonth,sDay,sHour,sMinute,sSecond)
}
var nowdate = sdf.format(nowdt.getTime)

def getDate(year:Int, month:Int, day:Int, add:Int) : String = {
  val resDate = "%tY/%<tm/%<td" format new java.util.GregorianCalendar(year, month-1, day-add);
  return resDate;
}

def zeroPadding2(str:String) : String = {
  var res = str
  if(str.length == 1)
    res = "0"+str
  return res
}

val nowdatesplit = nowdate.split("/")
val nowYear = nowdatesplit(0).toInt
val nowMonth = nowdatesplit(1).toInt
val nowDay = nowdatesplit(2).toInt
val nowHourStr = nowdatesplit(3)
val nowMinuteStr = nowdatesplit(4).substring(0,1)+"0"

//全データをmetricDfに格納
var metricDf:org.apache.spark.sql.DataFrame = null

try {

  for(addDay <- (0 to (dayRange-1))) {

    for(addHour <- ((24-1) to 0 by -1)) {

      for(addMinute <- List(50,40,30,20,10,0)) { //50,40,30,20,10,0

        var analysisDate = getDate(nowYear, nowMonth, nowDay, addDay)
        var fileDate = analysisDate.replaceAll("/","")
        var dateSplit = analysisDate.split("/")
        var year = dateSplit(0)
        var month = dateSplit(1)
        var day = dateSplit(2)

        var hourStr = zeroPadding2(addHour.toString)
        var minuteStr = zeroPadding2(addMinute.toString)
        var minutePath = minuteStr.substring(0,1)
        var minutePathForDisplay = minuteStr.substring(0,2)

        var metricDfMg = sqlContext.read.json(baseLocalPath+"SERVER-MG/"+year+"/"+month+"/"+day+"/metricbeat_"+fileDate+hourStr+minutePath+"*", "SERVER-MG")
        var metricDfS1 = sqlContext.read.json(baseLocalPath+"SERVER-SR01/"+year+"/"+month+"/"+day+"/metricbeat_"+fileDate+hourStr+minutePath+"*", "SERVER-SR01")
        var metricDfS2 = sqlContext.read.json(baseLocalPath+"SERVER-SR02/"+year+"/"+month+"/"+day+"/metricbeat_"+fileDate+hourStr+minutePath+"*", "SERVER-SR02")
        var metricDfS3 = sqlContext.read.json(baseLocalPath+"SERVER-SR03/"+year+"/"+month+"/"+day+"/metricbeat_"+fileDate+hourStr+minutePath+"*", "SERVER-SR03")

        val path = year+"/"+month+"/"+day+"/metricbeat_"+fileDate+hourStr+minutePathForDisplay
        println(path)

        if (metricDf == null) {
          metricDf = metricDfMg.unionAll(metricDfS1).unionAll(metricDfS2).unionAll(metricDfS3)
          metricDf = metricDfMg
        } else {
          metricDf = metricDf.unionAll(metricDfMg).unionAll(metricDfS1).unionAll(metricDfS2).unionAll(metricDfS3)
        }

        metricDfMg = null; metricDfS1 = null; metricDfS2 = null; metricDfS3 = null;

      }
    }
  }
} catch {
  case iie:java.io.IOException => println(iie)
}

DataFrameに対して、意味のある情報の抜き出し・データ整形といった処理を行う

今回取り出したい情報は、CPU利用率、ディスクI/O、ファイルシステム利用率、メモリ利用率、ネットワーク流量の5つです。
metricDfには余計な情報も含まれているので、これらの情報のみを抜き出します。
さらに、メトリクスには累計ネットワーク流量が記録されていたので、これを使って単位時間あたりのネットワーク流量を計算します。

まずは前の処理で得たmetricDfから上記の項目ごとの切り出しを行い、それぞれの項目ごとにDataFrameを作成します。

import sqlContext.implicits._
import org.apache.spark.rdd.RDD
import org.apache.spark.sql._
import org.apache.spark.sql.functions.{unix_timestamp, to_date}

//DataFrameから項目ごとに切り出し、それぞれの項目ごとにDataFrame作成
var cpuDf = metricDf.where("metricset.name like '%cpu%'")
var diskDf = metricDf.where("metricset.name like '%diskio%'")
var fsDf = metricDf.where("metricset.name like '%filesystem%'")
var memoryDf = metricDf.where("metricset.name like '%memory%'")
var networkDf = metricDf.where("metricset.name like '%network%'")

//必要なカラムだけ取り出す
cpuDf = cpuDf.select("@timestamp", "beat.hostname", "system.cpu.system.pct", "system.cpu.user.pct")
val newNamesCpu = Seq("Date", "Host", "CPU_Usage_By_System", "CPU_Usage_By_User") //パーセント表示
cpuDf = cpuDf.toDF(newNamesCpu:_*)

diskDf = diskDf.select("@timestamp", "beat.hostname", "system.diskio.read.bytes", "system.diskio.write.bytes")
val newNamesDisk = Seq("Date", "Host", "Disk_IO_Reading_Bytes", "Disk_IO_Writing_Bytes") //バイト表示
diskDf = diskDf.toDF(newNamesDisk:_*)

fsDf = fsDf.select("@timestamp", "beat.hostname", "system.filesystem.used.pct")
val newNamesFs = Seq("Date", "Host", "Filesystem_Usage") //パーセント表示
fsDf = fsDf.toDF(newNamesFs:_*)

memoryDf = memoryDf.select("@timestamp", "beat.hostname", "system.memory.used.pct")
val newNamesMemory = Seq("Date", "Host", "Memory_Usage") //パーセント表示
memoryDf = memoryDf.toDF(newNamesMemory:_*)

networkDf = networkDf.select("@timestamp", "beat.hostname", "system.network.in.bytes", "system.network.out.bytes")
val newNamesNetwork = Seq("Date", "Host", "Network_In_Bytes", "Network_Out_Bytes") //バイト表示
networkDf = networkDf.toDF(newNamesNetwork:_*)

このあと、DataFrameのままSQLライクなメソッドチェーンを繋げて処理をしていたのですが、ものすごく実行時間が掛かってしまっていました。
そこで、DataFrameをRDD経由でScalaのArrayに変換し、Arrayに対して処理を行うことで、処理速度が大幅に改善されました。

DataFrameをScalaのArrayとして操作する

CPU利用率についての処理を行います。
まず、Dataをパースし、最小時間単位をhourに直します。

//Dateをmapして、最小時間単位をhourにするよう部分文字列切り出し
val DateTime = ("""([0-9]{4}-[0-9]{2}-[0-9]{2})""" + "T" + """([0-9]{2}:[0-9]{2}:[0-9]{2}.[0-9]{3})""" + "Z").r

def get(s:String) = s match {
  case DateTime(d,t) => d + " " + t.slice(0,2)
}

var rowsCpu: RDD[Row] = cpuDf.rdd
var cpuDfMapped = rowsCpu.map({
  case Row(date: String, host: String, cPU_Usage_By_System: Double, cPU_Usage_By_User: Double) => (get(date), host, cPU_Usage_By_System, cPU_Usage_By_User)
}).toDF("Date", "Host", "CPU_Usage_By_System", "CPU_Usage_By_User")

1週間分のデータを可視化するので、毎時間最大値の情報のみを使うことにします。

import org.apache.spark.sql.functions.{lead, lag}
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._

//Cpuログをホストごとに分類 <- この部分、処理を軽くするために行っているので場合によっては分けなくてよい(groupByでDate,Hostを指定しているため)
var cpuDf_filtered_mg = cpuDfMapped.where('Host === "SERVER-MG")
var cpuDf_filtered_1 = cpuDfMapped.where('Host === "SERVER-SR01")
var cpuDf_filtered_2 = cpuDfMapped.where('Host === "SERVER-SR02")
var cpuDf_filtered_3 = cpuDfMapped.where('Host === "SERVER-SR03")

//毎時間最大値を抽出
//rowscpuDf_mgの型: Array[org.apache.spark.sql.Row]
var rowscpuDf_mg = cpuDf_filtered_mg.groupBy("Date", "Host").agg(max("CPU_Usage_By_System"), max("CPU_Usage_By_User")).sort($"Date").rdd.collect()
var rowscpuDf_1 = cpuDf_filtered_1.groupBy("Date", "Host").agg(max("CPU_Usage_By_System"), max("CPU_Usage_By_User")).sort($"Date").rdd.collect()
var rowscpuDf_2 = cpuDf_filtered_2.groupBy("Date", "Host").agg(max("CPU_Usage_By_System"), max("CPU_Usage_By_User")).sort($"Date").rdd.collect()
var rowscpuDf_3 = cpuDf_filtered_3.groupBy("Date", "Host").agg(max("CPU_Usage_By_System"), max("CPU_Usage_By_User")).sort($"Date").rdd.collect()

//cpuDf_mg: Array[(String, String, Double, Double)]
var cpuDf_mg = rowscpuDf_mg.map{u => (u.getString(0), u.getString(1), u.getDouble(2), u.getDouble(3)) }
var cpuDf_1 = rowscpuDf_1.map{u => (u.getString(0), u.getString(1), u.getDouble(2), u.getDouble(3)) }
var cpuDf_2 = rowscpuDf_2.map{u => (u.getString(0), u.getString(1), u.getDouble(2), u.getDouble(3)) }
var cpuDf_3 = rowscpuDf_3.map{u => (u.getString(0), u.getString(1), u.getDouble(2), u.getDouble(3)) }

//Arrayを統合し、DataFrameに戻しテーブル登録
var cpuDfs = cpuDf_mg ++ cpuDf_1 ++ cpuDf_2 ++ cpuDf_3
var cpuDfs_modified = sc.makeRDD(cpuDfs).toDF("Date", "Host", "CPU_Usage_By_System", "CPU_Usage_By_User")
cpuDfs_modified.registerTempTable("T_CPU")

ディスクI/O・ファイルシステム利用率・メモリ利用率の他3つのデータについての操作は、これと同様に行いました。

Scalaのコレクションメソッドmapを使った処理

ネットワーク流量についての処理を行います。
メトリクスに記録されていたネットワーク情報は、上り・下り転送量の累積バイト数であり、2^32-1バイトで循環的になっていました。
そこで、1つ前のデータとの差分を取ることで、単位時間あたりのデータ転送量を計算しました。

//Dateをmapして、最小時間単位をhourにするよう部分文字列切り出し
val DateTime = ("""([0-9]{4}-[0-9]{2}-[0-9]{2})""" + "T" + """([0-9]{2}:[0-9]{2}:[0-9]{2}.[0-9]{3})""" + "Z").r

def get(s:String) = s match {
  case DateTime(d,t) => d + " " + t.slice(0,2)
}

var rowsNetwork:RDD[Row] = networkDf.rdd
var networkDfMapped = rowsNetwork.map({
  case Row(date: String, host: String, network_In_Bytes: Long, network_Out_Bytes: Long) => (get(date), host, network_In_Bytes, network_Out_Bytes)
}).toDF("Date", "Host", "Network_In_Bytes", "Network_Out_Bytes")

Scalaのコレクションのメソッドであるmapを使って差分の計算を行います。負になった場合は循環の切れ目が入っていると判断し、カウンタ値2^32 - 1を使って処理しました。
SERVER-MGサーバのデータだけであとの3台分は省略しています。

import org.apache.spark.sql.functions.{lead, lag}
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._

//Networkログをホストごとに分類
var networkDf_filtered_mg = networkDfMapped.where('Host === "SERVER-MG")

//毎時間最大値を抽出 
//rowsNetwork_mg: Array[org.apache.spark.sql.Row]
var rowsNetwork_mg = networkDf_filtered_mg.groupBy("Date", "Host").agg(max("Network_In_Bytes"), max("Network_Out_Bytes")).sort($"Date").rdd.collect()

//差分を計算するための関数
def delta(a:Any, b:Any): Long = {
  var atmp: Long = a.asInstanceOf[Long]
  var btmp: Long = b.asInstanceOf[Long]
  if (atmp - btmp > 0) {
    return atmp - btmp
  } else {
    return 4294967295L - btmp + atmp //4294967295Lはカウンタ値
  }
}

//tailで先頭の要素を除いたArrayを、zipでもとのArrayの同インデックス要素とのタプルにし、mapで差分を計算
var rowsNetwork_mg_delta = rowsNetwork_mg.zip(rowsNetwork_mg.tail).map{ u =>
  (u._2.apply(0).toString,
  u._2.apply(1).toString,
  delta(u._2.apply(2), u._1.apply(2)),
  delta(u._2.apply(3), u._1.apply(3)))
}

//DataFrameに戻し、列の名前を付ける
val networkDf_mg_modified = sc.makeRDD(rowsNetwork_mg_delta).toDF("Date", "Host", "Network_In_Per_DeltaTime", "Network_Out_Per_DeltaTime")

//テーブル登録
networkDf_mg_modified.registerTempTable("T_NETWORK_MG")

SQL文を記述してグラフ出力

SQL文を書くだけで可視化してくれます。

%sql
select
  *
from
  T_CPU
order by
  Date

GUI操作で、使用する列やグラフの形状を変えることができます。
結果はこのようになりました。

f:id:brains-tech:20170425160512p:plain

SORACOM Funnel ときどき 異常検知

本記事はSORACOMサービスリリース1周年記念ブログの11月2日分です。



皆さん、こんにちは。ブレインズテクノロジーの林です。
今日はSORACOMリリース1周年のブログに寄稿させていただくことになり、以前から気になっていたSORACOM Funnelに関する記事を書かせていただきます。

KYOSOの辻様の記事(RaspberryPi +SORACOM FunnelでIoTデータを閉域網でS3やDynamoDBに格納する)ですでに詳しく説明いただたいておりますので、重複する部分は割愛して、SORACOM Funnelを利用するメリット、データを活用する部分にフォーカスして寄稿いたします。

SORACOM Funnel を利用するメリット

SORACOM Funnelを利用しない場合の、IoTデータの収集〜可視化するまでの流れは、以下の通りです。各デバイスからエージェントを介して、AWS IoT や 独自のGatewayへデータを送信し、ハブとしてKinesisを利用し、各データアプリケーションが可視化・分析等を行うのが一般的な構成になるかと思います。

f:id:brains-tech:20161102022443p:plain

が、しかし、SORACOM Funnelを利用すれば、アーキテクチャーで考慮しなければならないポイントがかなり減ります。

  • 通信経路・データセキュリティに関する考慮(閉域網での接続)
  • データを収集する部分の可用性・拡張性に対する考慮(AWS IoTであれば問題ないが・・・)
  • バイスと収集層との通信プロトコルに関する検討(mqtt / tcp / udp ?)

f:id:brains-tech:20161102022451p:plain

他にも勿論あると思いますし、SORACOMだからこそ可能な、SIM単位での通信ON/OFFの制御、アクセスコントロールの割当の容易さなど、メリットをあげればキリがないですね。つまり、異常検知・データ分析を主戦場とする我々とデバイスの距離が縮まったような気がします。

というわけで、SORACOM Funnel を利用してデバイスからデータを収集し、データの特性分析を行い、異常検知を行う流れをやってみたいと思います。

SORACOM Funnel を介した 異常検知

今回は、RaspberryPi上にセンサーデータが集まるという過程で、SORACOM Air → SORACOM Funnelを経由して、Kinesisへ流れているデータをImpulseが蓄積、分析、リアルタイムの異常検知をかけていくシナリオです。

f:id:brains-tech:20161102030130p:plain

※ Device〜Kinesisへの設定方法は、前述のKYOSOの辻様の記事(RaspberryPi +SORACOM FunnelでIoTデータを閉域網でS3やDynamoDBに格納する)をご参照ください。


複数個の照度センサーデータを一定間隔:5分でデータを送信し、収集したデータ特性を分析し、相関崩れによる異常検知を行うまでの流れになります。照度センサー4つを時系列でほぼ同じような値を取ることを前提として、データ送信をしています。(夜間帯は低く、日中にかけて照度の値が大きくなるような形)

まずは、RaspBerryPi から以下のデータをSORACOM Funnel のエンドポイントに対してデータを送信します。

{
    "description": "illuminance_data"
    , "illuminance_4": 3.88
    , "illuminance_3": 3.67
    , "illuminance_2": 4.63
    , "illuminance_1": 4.59
    , "time": 1477752900
}

そうすると、Kinesisへ以下の通りデータが入ってきます。
各種情報が付与され、送り出したデータは payloads に入っています。
便利ですね!!

{
    "operatorId": "xxxxxxxxxxxx"
   ,"timestamp": 1478055514828
   ,"destination": {
       "resourceUrl": "https://kinesis.us-west-2.amazonaws.com/soracom_funnel"
      ,"service": "kinesis"
      ,"provider": "aws"
    }
    ,"credentialsId": "funnel_kinesis"
    ,"payloads":{
        "description": "illuminance_data"
       ,"illuminance_4": 3.88
       ,"illuminance_3": 3.67
       ,"illuminance_2": 4.63
       ,"illuminance_1": 4.59
       ,"time": 1477752900
     }
     ,"sourceProtocol": "http"
     ,"imsi": "9999999999999999"
}

さて、Kinesisに流されたデータをImpulseへ取り込み、データの特性を分析します。図の右側の通り、特性分析を行うと、照度センサー1〜4の値は非常に高い相関を示していることがわかります。

f:id:brains-tech:20161102134833p:plain

収集データから異常検知を行うために、シュミレーションを行って初期モデルを作成します。作成されたモデルで、Impulseのリアルタイム検知機能を有効にします。有効にした後、RaspberryPiから流すデータの照度を一つだけ小さくすると、相関が崩れたとImpulseが判定し、"異常(いつもと違う)"と判定てくれます。

f:id:brains-tech:20161102134839p:plain

まとめ

SORACOM / SORACOM Funnelを利用することで、本来やりたいことに注力できるようになります。
つまり、データの分析・活用という部分に注力することができるようになるわけです。今後も継続して、新しい技術をウォッチしていこうと思います。

ビバ、SORACOM

おまけ

今回使ったRaspberryPi + SORACOM Sim のセットの写真です。
f:id:brains-tech:20161102162950j:plain

ET/IoT総合技術展・関西 出展レポート

こんにちは!マーケティング担当の安部です。

今日8月8日はブレインズテクノロジーの創立記念日
早いもので9年目に突入しました…!
ブログをご覧の皆様、日々応援してくださっている皆様、ありがとうございます。
「最先端のオープンテクノロジーで、エンタープライズの世界に技術革新を提供する」明るい未来を創造する技術集団を、引き続きよろしくお願いします!

 

さて、今回のブログは先日の投稿に続き出展レポートです。

当社は2016年7月7日(木)~7月8日(金)にグランフロント大阪で行われた「ET/oT総合技術展・関西」にAWSさんのブースから出展しました。
(オージス総研さん、東海ソフトさんともご一緒させていただきました。)

ブレインズ初の関西出展です!

 

クラウド×機械学習×異常検知

今回も、デモを中心に異常検知のソリューションをご紹介しました。

f:id:brains-tech:20160726143153j:plain

前回のInteropと比べ、来場者の方とブースでじっくりお話しする時間が多かったです。「IoT総合技術展」の名のとおり、IoTの実用的なソリューションを求めている企業が多いことを改めて実感できました。

来場者の方だけでなく、出展企業の方々からもImpulseの機械学習を使った異常検知を高く評価していただき、ビジネスアライアンスの検討に繋がったことも大きな収穫でした。 

FA機器への対応についての問い合わせがあった際、隣のブースの東海ソフトさん(写真では見切れていますが…)と連携して一緒にお話しすることもありました。

f:id:brains-tech:20160726143321j:plain

また、今回はブースでのご紹介以外に「AWSで実現する IoT向け故障予兆検知・分析サービス 『Impulse』のご紹介 」というタイトルでセッションも行いました。
立ち見にも関わらずたくさんの方が聴いてくださり、なんと2日連続でボーナスセッションをゲットしました。

f:id:brains-tech:20160726143218j:plain

 最後に

今回も連日100名近いお客さまにご来場頂き大盛況で終えることができました。
改ましてご来場頂きました皆様、出展の機会を頂いたAWSの皆様、共同出展でお世話になった皆様、ありがとうございました!

Impulseの次回出展は…ご期待ください! 

Interop Tokyo 2016 出展レポート

こんにちは!マーケティング担当の安部です。

当社は2016年6月8日(水)~6月10日(金)幕張メッセで行われたInterop Tokyo 2016に出展いたしました。
連日100名を超えるお客様にご来場いただき、当社製品「Impulse」をご紹介することができました。ありがとうございました!

イベントが終了して早くも一ヶ月が経とうとしていますね…

レポートが遅くなってスミマセン。

f:id:brains-tech:20160701192406j:plain

当社のブースは、IoT Worldのエリアに出展しました。

リアルタイム予測分析プラットフォーム Impulse

「Impulse」は、リアルタイムの分析プラットフォームです。
*詳細はこちらをどうぞ!
Impulse(リアルタイム大規模データ分析基盤) | 製品&サービス | ブレインズテクノロジー株式会社

今回は、IoT用途として「照度センサーを使った異常検知」と「不良品検出ソリューション」のデモを紹介しました。

f:id:brains-tech:20160701192521j:plain

Best of Show Award」特別賞受賞

そして・・・突如舞い込んだビッグニュース。

300近くエントリーされた新製品の中、また、名だたる企業がファイナリストに選ばれる中…まさかの受賞でした。
初出展の企業がアワードを受賞するのは、初めての出来事だそうです。

f:id:brains-tech:20160701192527j:plain f:id:brains-tech:20160701192814j:plain


審査員の方々からは、

- 機械学習を実用性の高い分野でサービスとして展開していること

- 利用者のメリットが明確であること

を高く評価して頂きました。


f:id:brains-tech:20160701192448j:plain

ブースを訪れた方々からは、

機械学習ってよく聞くけど、結局良くわからない…」
「なんとなく興味はあるけど、実際どう使われているの?」

というコメントを多く耳にしました。

私たちはImpulseを「異常検知」という実用性の高い分野でサービス展開し、多くの企業の皆さまが機械学習活用の恩恵を享受できる努力を続けていきたいと思っています。

 

最後に

来場してくれた方のブログに・・・Impulseが紹介されていました。
Interop2016で見つけたヤンデレ彼女用IoTまとめ [Interop Tokyo 2016] | ツチノコブログ


今後も様々なイベントへ出展し、異常検知に関わるソリューションを紹介していきます。Impulseの「これから」にご期待ください!

機械学習アルゴリズム実装シリーズ vol 2[ロジスティック回帰編]

こんにちは。ブレインズテクノロジー樋口です。
機械学習アルゴリズムを一から実装するシリーズ2回目。今回のお題はロジスティック回帰(Logistic Regression)の実装。
前回と同様あまり難しい説明はなしに実装していくつもりですが、もっと詳しいこと教えんかい( ゚д゚ )クワッ!!となったら以下のような専門書をどうぞ。
machinelearningmastery.com

ロジスティック回帰とは

ロジスティック回帰(Logistic Regression)は分類問題を解決するための手法で教師あり学習に分類されます。
元々は統計の分野で発展してきたもので、応用先としては
・e-mailのスパム判定
・病理標本の悪性判定
トランザクションの正常性判定

などに使われているようです。
# 名前には回帰(Regression)とついてるんですけど分類なんですね。。。

ロジスティック回帰[理論編]

前回、線形回帰ではモデルを以下のような数式で表しました。
h_θ(x)=θ_0+θ_1x
ロジスティック回帰の場合は上記の式を更にシグモイド関数という関数に入れます。式で表わすと以下のようになります。
h_θ(x)=g(θ^TX) -①
g_θ(z)=\frac{1}{1+e^{-z}}
※ちみなにシグモイド関数は以下のような波形を描きます。
f:id:brains-tech:20160307110209p:plain
シグモイド関数 - Wikipedia
jp.mathworks.com



①により、データを入力すると0~1の間の値を返してきて、それがあるクラスに当てはまる確率となります。
例えば、ある特徴量x_0, x_1を入力した時、出力された値がh_θ(X)=0.7だった場合、それが当該クラスである確率は70%であるといった具合です。

では、①の式が適切な値を出力してくれるようにθの値を学習していくためにはどうすればよいでしょうか。

前回と同じく、目的関数を定めた後その目的関数を最小とするような θの値を求めていきます。

ロジスティック回帰では目的関数に以下の式を使用します。
J(θ_0, θ_1)=-\frac{1}{m}\sum_{i=1}^{m} (y^i\log(h_θ(x^i)) + (1-y^i)\log(1-h_θ(x^i)) -②
xは特徴量, yはラベル

ここからは前回と同じです。最急降下法を使って、このJ(θ_0, θ_1)を最小にする\thetaを求めます。
ここで偏微分とかいろいろ出てくるのですが、とりあえず置いておき、つまるところ以下の式を計算していくことになります。
次のθ_j=θ_j-α\sum_{i=1}^{m} ((h_θ(x^i)-y^i)x^i_j) -③

データを見る

以上がかなり簡単な理論のお話です。
移行は実装に入っていきますが、まずは今回のデータを見てみましょう。
以下の様なデータを用意しました。
f:id:brains-tech:20160403195901p:plain

赤、青でプロットされた点があり、それぞれがクラス(赤: class 1, 青: class 0)を表しています。
今回はこの2つのクラスのデータを学習して、新しい点が入力された時そのデータが赤クラスである確率はどれくらいなのか、つまりどちらのクラスなのか分類したいと思います。
※実装がメインですので、シンプルなデータにしてあります。

実装!

ここまで説明してきたことをコードに落とすと以下のようになります。
コメントで説明を入れてありますが、分かりにくいところは少し動かしてみることをオススメします。
[コード]

# -*- coding:utf-8 -*-
import numpy as np
import matplotlib.pyplot as plt


def gradient_decent(theta, X, y, iter_num, alpha):
    """
    最急降下法メソッド
    与えられたiter_num分だけ最急降下法を適用してJ(θ)を最小とするthetaを求める
    :param theta: モデルのパラメータθ
    :param X: 入力データ(学習に使われるデータ)
    :param y: 入力データに対するラベル
    :param iter_num: 最急降下法の適用回数
    :param alpha: 学習率
    :return: theta, J (学習後のtheta, 学習途中での目的関数の値)
    """

    m = float(len(X))
    J = []
    for i in range(iter_num):
        # 現在のthetaでh_thetaを計算 (
        h = sigmoid(np.inner(theta, X))  # --- ①式
        # 目的関数を可視化のために計算
        J.append(1/m * np.sum((-y*np.log(h) - (1-y)*np.log(1-h))))  # --- ②式
        # 以下で, ③式を計算してthetaを更新していく
        theta1_tmp = 1/m * np.sum((h-y)*X[:, 0])
        theta2_tmp = 1/m * np.sum((h-y)*X[:, 1])
        theta3_tmp = 1/m * np.sum((h-y)*X[:, 2])
        theta = theta - alpha*np.array([theta1_tmp, theta2_tmp, theta3_tmp])
        print "theta: ", theta
    return theta, J


def sigmoid(x):
    """
    与えられたデータをsigmoid関数へ入れて計算結果を返す
    sigmoid関数
    :param x: 入力データ
    :return: sigmoid関数適用後のデータ
    """
    return 1/(1 + np.exp(-x))
    

def main():
    # サンプルデータのロード
    # 前述した散布図のデータ
    sample_data = np.load("./data.npy")

    # ラベルデータのロード
    # 赤: 1, 青: 0 として表現
    y = np.load("./label.npy")

    # 学習データの作成
    x_tmp = np.ones(len(sample_data))
    X = np.c_[x_tmp, sample_data]

    # パラメータの初期化
    # thataは任意の整数
    # alphaは学習率で予めこちらで調整した値が代入されています
    # iter_numの回数により何回最急降下法が回るかが決まります
    theta = np.array([1,1,1])
    alpha = 0.5
    iter_num = 100

    # 最急降下法を使って学習を行う
    theta, J = gradient_decent(theta, X, y, iter_num, alpha)
    plt.plot(J)
    plt.show()


if __name__ == "__main__":
    main()


[実行結果]

$ python lr.py
theta:  [ 0.8319953   0.68474584  0.81423634]
theta:  [ 0.67634834  0.38081935  0.63610057]
theta:  [ 0.53901998  0.09768263  0.47012794]
theta:  [ 0.42658727 -0.1536833   0.32244438]
theta:  [ 0.34236108 -0.36727612  0.19797763]
theta:  [ 0.28391004 -0.54596135  0.09699565]
theta:  [ 0.24549959 -0.69715343  0.01567327]
theta:  [ 0.22146887 -0.82779591 -0.05084846]
theta:  [ 0.20750074 -0.94295183 -0.10667272]
theta:  [ 0.6395262  -2.837131   -1.09546083]
...
...
...
theta:  [ 0.89455876 -3.43082868 -1.50299413]
theta:  [ 0.90015444 -3.44389053 -1.51225235]
theta:  [ 0.90569562 -3.45683949 -1.52143847]
theta:  [ 0.91118314 -3.46967788 -1.53055366]
theta:  [ 0.91661788 -3.48240793 -1.53959907]
theta:  [ 0.92200064 -3.49503179 -1.54857581]
theta:  [ 0.92733226 -3.50755156 -1.55748496]
theta:  [ 0.93261354 -3.51996927 -1.56632758]

[目的関数J(θ)の推移]
f:id:mhigu:20160314074359p:plain

学習の結果、目的関数の値が小さくなり収束していっていることが分かります。

確認のために学習したモデルに値を入力して結果を観測します。
以下にテスト用に別に作ったスクリプトを記載します。
x_0=-2, x_1=1の場合

# -*- coding:utf-8 -*-

import numpy as np
import matplotlib.pyplot as plt


def sigmoid(x):
    # 与えられたデータをシグモイド関数へ入れて
    # 計算された結果を返す
    return 1/(1 + np.exp(-x))
    

def test():

    # プロット用にデータをロード
    X = np.load("data.npy")
    label = np.load("label.npy")

    # theta:  [ 0.93261354 -3.51996927 -1.56632758] を利用
    theta = np.array([0.93261354, -3.51996927, -1.56632578])

    # x0=2, x1=1を設定
    x = np.array([1, -2, 1])
    class_1_rate = sigmoid(np.dot(theta, x))
    class_num = -1

    # モデルから返ってきた値が0.5より大きければ赤(class 1)
    # 0.5以下であれば青(class 0) 
    if class_1_rate > 0.5:
        class_num = 1
    elif class_1_rate <= 0.5:
        class_num = 0
    print "[RESULT]"
    print "################"
    print "Model Output: {0}".format(class_1_rate)
    print "Input Data is class {0}".format(class_num)
    print "################"

    # 散布図をプロット
    plt.scatter(X[:,0], X[:,1], c = label, s= 100)
    plt.scatter(x[1], x[2], marker='^', color='g', s=150)
    plt.show()

if __name__ == "__main__":
    test()

[実行結果]

[RESULT]
################
Model Output: 0.998351478053
Input Data is class 1
################

ちゃんと動いているように見えます。。。
散布図を作って見るとしっかり分類できていますね。(緑の三角がinputしたものです)
f:id:brains-tech:20160404000807p:plain
ひとまずは実装完了です。
が、本来であれば、過学習を考慮して罰則項を足してモデルを作ったり、特徴量を取るときに正規化を行ったりと様々工夫が入ります。興味のある方はそのへんまで是非是非調べてみてください。

結び

今回は分類問題を解くための教師あり機械学習アルゴリズム、ロジスティック回帰を実装してきました。
分類問題を解くための機械学習アルゴリズムには、ロジスティック回帰の他にも、SVMニューラルネットワークなどたくさんあります。最近ディープラーニングがすごく話題になってますが、ニューラルネットワークはその走りですよね。次回は、そのニューラルネットワークを今回のロジスティック回帰との違いに着目しながら実装していきたいと思います。

ではでは(((((((((((っ・ω・)っ ブーン

Spark1.6.0のDataset APIを触ってみた

あけましておめでとうございます。 Impulse開発チームの木村です。

今回は、Spark 1.6.0で導入されたDataset APIを、spark-shell上で触ってみました。

Dataset APIとは

Dataset APIは、RDDやDataFrameと同じく、データのまとまりを扱うためのAPIです。 RDDとDataFrame双方の長所を合わせ持つAPIとして、開発が進められています。 (なお、ver1.6.0でのDataset API導入は、あくまで実験的なものであり、ver2.0.0での本リリースが予定されているようです。)

対応するIssueによれば、Datasetの要件として次が掲げられています。

  • Fast
    • ほとんどのケースで、RDD以上のパフォーマンス
  • Typesafe
  • Support for a variety of object models
    • デフォルトで様々なオブジェクトを、Datasetの要素としてエンコード可能
  • Java Compatible
  • Interoperates with DataFrames
    • DatasetとDataFrame間での(お決まりのコードを書く必要のない)シームレスな変換

ver1.6.0時点では、ScalaJavaからDataset APIを使用できます。 Pythonのサポートは、先のリリースとなるそうです。

Datasetの作り方

  • Seqから作る

toDSを使って簡単に、SeqをDatasetに変換できます。

// spark-shell上では、sqlContext.implicits._のimportは不要

// 数が要素のDataset
scala> val numberDS = Seq(1, 2, 3).toDS()
ds: org.apache.spark.sql.Dataset[Int] = [value: int]

scala> numberDS.show()
+-----+
|value|
+-----+
|    1|
|    2|
|    3|
+-----+

// case classのインスタンスが要素のDataset
scala> :paste
case class Player(
  name: String,
  weapon: String,
  rank: Int
)
defined class Player

scala> :paste
val playerDS = Seq(
  Player("Ponce", "Splattershot Jr.", 9), 
  Player("Paciorek", "E-Liter 3K", 50),
  Player("Petrick", "Splattershot Jr.", 12)
).toDS()
playerDS: org.apache.spark.sql.Dataset[Player] = [name: string, weapon: string, rank: int]

scala> playerDS.show()
+--------+----------------+----+
|    name|          weapon|rank|
+--------+----------------+----+
|   Ponce|Splattershot Jr.|   9|
|Paciorek|      E-Liter 3K|  50|
| Petrick|Splattershot Jr.|  12|
+--------+----------------+----+
  • DataFrameから作る

DataFrame#as[T]を使って、DataFrameをDatasetに変換できます。

// 適当なDataFrameを作成
scala> val df = sqlContext.read.json(sc.parallelize("""{"x": 11, "y": 22}""" :: Nil))
df: org.apache.spark.sql.DataFrame = [x: bigint, y: bigint]

scala> case class Coordinate(x: Long, y: Long)
defined class Coordinate

// DataFrameのカラムとcase classのフィールドの対応は、名前で判断される
scala> val ds = df.as[Coordinate]
ds: org.apache.spark.sql.Dataset[Coordinate] = [x: bigint, y: bigint]

scala> ds.show()
+---+---+
|  x|  y|
+---+---+
| 11| 22|
+---+---+

Datasetへの操作

今までと似た感覚で操作を行えます。

  • フィルタ
scala> val numberDS = Seq(1, 2, 3).toDS()
numberDS: org.apache.spark.sql.Dataset[Int] = [value: int]

scala> numberDS.show()
+-----+
|value|
+-----+
|    1|
|    2|
|    3|
+-----+

scala> :paste
numberDS
  .filter(_ >= 2)
  .show()
+-----+
|value|
+-----+
|    2|
|    3|
+-----+
  • 集計
scala> :paste
case class Player(
  name: String,
  weapon: String,
  rank: Int
)
defined class Player

scala> :paste
val playerDS = Seq(
  Player("Ponce", "Splattershot Jr.", 9), 
  Player("Paciorek", "E-Liter 3K", 50),
  Player("Petrick", "Splattershot Jr.", 12)
).toDS()
playerDS: org.apache.spark.sql.Dataset[Player] = [name: string, weapon: string, rank: int]

scala> playerDS.show()
+--------+----------------+----+
|    name|          weapon|rank|
+--------+----------------+----+
|   Ponce|Splattershot Jr.|   9|
|Paciorek|      E-Liter 3K|  50|
| Petrick|Splattershot Jr.|  12|
+--------+----------------+----+

scala> :paste
playerDS
  .groupBy(_.weapon)
  .agg(count("weapon"))
  .show()
+----------------+-------------+
|           value|count(weapon)|
+----------------+-------------+
|      E-Liter 3K|            1|
|Splattershot Jr.|            2|
+----------------+-------------+

ただし、RDDやDataFrameに存在する操作の一部(orderBydropなど)は、現時点では実装されていません。 このような操作を行う場合は、一度RDDやDataFrameに変換する必要があります。 (Dataset#rddRDDに、Dataset#toDFでDataFrameに変換することができます)

おわりに

Dataset APIについて紹介しました。

Sparkはバージョンアップの間隔が短く、新しい機能がどんどん出てきます。 追っていくのは比較的大変ですが、良い新機能はどんどん取り入れるべく、精進していこうと思います。

(deprecatedが頻発しないといいなあ…)

参考

World of IoTに出展したデモ

こんにちは。おじさんチームのこんのです。
今回は、去る12月16日〜18日に東京ビッグサイトで開催されたWorld of IoT | SEMICON Japanで出展したデモについて紹介したいと思います。
なお、この3日間のイベントで、200名を超える方々が弊社のImpulseについて興味を持ち、お話を聞きにきてくださいました。ありがとうございます!!

デモは、弊社サービスのひとつである、リアルタイム大規模データ分析基盤Impulseを使った異常検知です。
Impulseとはなんぞや?という方は下記ページをご覧ください。
www.brains-tech.co.jp

太陽光発電パネルの発電量の故障予兆検知を想定し、それに似せた構成を取りました。
なお、今回のデモにあたり、アットマークテクノさんからArmadillo-IoTをお貸しいただきました。
アットマークテクノさんのご厚意に、この場を借りてお礼申し上げます!

f:id:bti-konno:20151218230257p:plain

概要構成は上記図のとおりです。ちょっとだけ詳しく流れを説明すると、

  1. 4つ並べた照度センサー(太陽光発電パネルの代わり)で照度(発電量の代わり)を測定し、その照度をアットマークテクノさんのArmadillo-IoTで収集します。
  2. 照度センサーから受け取ったデータは、アットマークテクノさんに作成いただいたプログラムにてファイルにJSON形式で追記されていきます。
  3. 上記ファイルを fluentd で監視し、6秒おきに溜まったデータを MQTT でAWS IoTにPublishします。
  4. AWS IoT から弊社Impulseにデータを流し込みます。
  5. Impulseでは正常状態を学習させておき、学習したモデルから入力されたデータが大きく外れていないかを検知します。
  6. 異常を検知した場合、AWS IoTのRasberry Pi2のShadowを書き換えます。
  7. Raspberry Pi2 は、Shadowが書き換わっていた場合、Arduino、LED、スピーカーを使って異常を知らせます。

Armadillo-IoTは、開発環境で使われるDebian(SDカード)で起動するようにしています。開発環境DebianでArmadillo-IoTを使用するとき、SWAP領域を作ることを忘れないでください。これがないとC/C++のソースからのコンパイルがかなりの割合で失敗します。ご注意ください。
また、dateが2000年スタートになっています。気付かずに証明書を入れてMQTTSで通信しようとするとSSLエラーがでます。日付時刻修正をお忘れなく。

fluentdからAWS IoTにMQTTでPublishするPluginは、yuuna/fluent-plugin-mqtt · GitHub を使わせていただきました。
yuunaさんにこの場を借りてお礼を申し上げます!

今回SORACOMさんのモバイルネットワークを使ってAWS IoTへ連携していますが、SORACOM Beamは使っていません。
# SORACOMさんの回線を使うことに決めたのがイベント前日ということもあり・・・。
# つまり、SORACOMさんのセットアップはものすごく簡単ということですね、はい。
# Armadillo-IoT、Raspberry Pi両方ともSORACOMさん接続です。
# Armadillo-IoTの3Gセットアップはアットマークテクノさんのマニュアルページに丁寧に記載されています。
# Raspberry Pi2は dietposterさんのQiitaを参考にしました。
# dietposterさん、ありがとうございます m(_"_)m

AWS IoTでは、PublishされたトピックのメッセージをそのままKinesisに流し込んでいます。

ImpulseではこのKinesisから流し込まれた値を取り込み、正常状態から外れていないかを指定された秒数間隔でチェックします。今回は同じような環境にある複数の照度センサーでしたので、Impulseのデータ特性分析から相関性が高い状態にあることがわかっています。つまり、Impulseではこの相関性が崩れた時に異常と判定します。
異常判定時は、Lambdaを通してRaspberry Pi2のShadowを書き換えています。

Raspberry Pi2では、AWS IoT SDK(node.js版)を使って、Shadowが書き換わった際に呼び出される function で(GPIO経由で)LEDを点灯させ、ミニディスプレイ付きArduino(実際には SparkFun Electronics のMicroView)にてメッセージを表示し、スピーカーから異常検知したことを伝えるようにしています。
f:id:bti-konno:20151219001132p:plain
スピーカーからお知らせする内容は、OpenJTalk を使って予め合成音声(合成音声には名工大のメイちゃんを使わせていただきました)を作っておいたものをaplayで再生しています。
なお、node.js(Javascript)からGPIOへ簡単にアクセスするために、Intelが開発している mraa ライブラリをインストールしています。mraa はIntel Edison等で主に使われていますが、Raspberry Piもカバー範囲に入っています。おすすめです。また、実際のコードは、AWS IoT Handsonで公開しているサンプルプログラムを若干改良しただけで済んでいます。

ざっくりと簡単に紹介いたしました。
いかがでしたか?
このような簡単な準備、構成で Impulse による異常検知ができてしまいます。

イベントの期間中、デモをお見せしたお客さまからこんなことを言われました。
「人がやっても同じではないか?Impulseは何がすごいのか?」
お客さま、いいことをおっしゃってくれました。それです。それなんです。
機械学習を活用することで、[人と同じこと]がコンピューター上でできるようになったんです。
さらにコンピューターでやることで、[人以上]に処理することができるのです。
確かに10個やその程度の数であれば人でやったほうが早いかもしれません。それが数百、数千、数万、それ以上あった場合はどうでしょう?膨大な数のデータソースを監視し異常を見つけ出すのは、人では途方も無いコストをかけなければ難しいでしょう。あるいは、それを長期にわたって絶え間なく監視し続けなければならないとしたらどうでしょう?やはり人でやるのは難しいと思います。人と同じように異常を見つけ出し、人以上に(長期間、膨大な数を)処理できる。それこそがImpulse、といいますか機械学習を活用したサービスが提供できるメリットなのです。

センサーでこれから何かしらデータを集めようとしているみなさま、分析しようとしているみなさま、故障予兆検知をしたいというニーズをお持ちのみなさま、Impulseであれば収集・蓄積・可視化から異常検知まで一通りの機能を提供できます。ぜひブレインズテクノロジーにご相談ください。
きっとお役に立てると思います。