Technology Topics by Brains

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

Spark Summit 2017 San Francisco

f:id:brains_aoki:20170607131148j:plain

こんにちは、データアナリストの青木とエンジニアの樋口です。

引き続き、Spark Summit 2017 San Fransiscoの記事です。Keynoteやセッションで特に興味深かったものを紹介していきます。

Keynote

Coming in Spark 2.2

まずは、Spark2.2に関する情報がきました。注目点は以下。

  • コストベースSQLの最適化

  • structured streamingがproduction-readyとなった

  • pip install pyspark が可能となる

すでにgitではv.2.2.0-rc4のtagが打たれていることから、リリース間近なようですね。

続いて大きな発表がありました。

Two new open source from Databricks

おそらくこの発表がSummitの目玉だったようです。 今後Databricksは以下2つについて、特に力を入れていくとのこと。

  1. Deep Learning

  2. Streaming Performance

Deep Learning

みんな大好きDeep Learning。画像分野を中心に華やかな成果を上げてはいますが、TensorFlowやKrasなど既存のDeep LearningオープンソースはまだまだAPIとしてlow-level。これに対してDatabricksはDeep Learning Pipelinesというhigh-levelなAPIを用意するぜ!とのことでした。これにより、

  • よくあるユースケースについては、コード数行でモデルを作成できる

  • Spark上で自動的に分散処理してくれる

  • バッチやストリーミングアプリ、SparkSQLにモデルをデプロイできる

ができるようになるとのことです。以下のDatabricksの記事に詳しい説明があります。

databricks.com

Streaming Performance

Spark Streamingにおけるデータに対し、DataFrameやSparkSQLなど、バッチと同様のhigh-levelのAPIを提供しつつ、レイテンシーを抑えるとのこと。microbatchの処理はやめて、continuous progressという新たな処理仕様になるようですが、既存のコード自体は変える必要はないようです。リアルタイム処理にSpark Streamingをコアとしている弊社にとってはうれしい話かもしれません。

Streaming PerformanceについてもDatabricksの以下の記事が詳しいです。

databricks.com

www.youtube.com

そのほか、DatabricksのServerlessの話もありました。

databricks.com

www.youtube.com

次はセッション。 各セッションの動画については来週末くらいにuploadされるみたいなので、その際にまたlinkと共に更新します。

Sessions

NEXT GENERATION WORKSHOP CAR DIAGNOSTICS AT BMW POWERED BY APACHE SPARK

f:id:mhigu:20170608101515p:plain:w160 f:id:mhigu:20170608101816p:plain:w160 f:id:mhigu:20170608101638p:plain:w160

概要

現在の販売店での車両診断は、手動で生成された決定木(人が作って来た条件分岐)に基づいていて、車種の多様化、車両システムの複雑化(hybrid, connective)に伴い、すでに限界に達しているとのこと。BMWでは、自動車やワークショップから入手できるデータを利用して、交換するべき部品や取るべき行動を予測できるモデルを作成し、そこから得られた結果をWeb-Applicationとして提供して、複雑化した車両整備を現場の人だけで解決できるような形にして出力しているそうです。

感想

これこそ、IoTの事例といった感じでしょうか。このアプリケーションを作り始めた理由としては、概要にも少し書いてありますが、車両システムの複雑化により整備者の知識だけでは交換すべき部品が分からないという背景があったそうです。この傾向は車業界に限らず発生することだと思うので、似たような事例が今後も出てくることになりそうです。impulseは異常検知に特化したソフトウェアとして作っているため、もしかしたら、impulseでも、、、?

DEEP LEARNING IN SECURITY—AN EMPIRICAL EXAMPLE IN USER AND ENTITY BEHAVIOR ANALYTICS (UEBA)

f:id:mhigu:20170608100119j:plain:w160 f:id:mhigu:20170608095516j:plain:w160 f:id:mhigu:20170608095526j:plain:w160

概要

このプレゼンテーションでは、顧客の攻撃検出例を使ってDeep Learningがどのように適用されて、どうセキュリティ問題を解決したのか、経験の共有、また、ディープラーニングや一般的な機械学習をより広範なセキュリティに展開するための課題とガイドラインについても説明していました。Deep Learningについては、以下の2つの実例が挙げられていました。

  • 畳み込みニューラルネットワーク(CNN)を使用するユーザ行動異常検出ソリューション
  • LSTMを使用するステートフルなユーザーリスクスコアリングシステム

感想

セキュリティ業界では、ルールベースの検知が限界とされて久しいですが、今回のプレゼンテーションではDeep Learning(CNN, LSTM)を使って異常検知をした実例を説明してくれました。CNNへの入力は、ユーザの通信状態(プロトコル毎)の画像。出力はどのユーザかのid。idがいつもと違うものになれば異常とする算段なのですが、これだけだと検知精度は良くないらしいです。対策として、CNNと併せてLSTMへ、フィッシングメールがあるかどうか、不審なDNSクエリがあるか、大量データ送信があるかなどをエンコードしたベクトルを入力して各段階でスコアを算出し、実際の検知を行っているとのこと。また、プレゼンの最後にセキュリティの異常検知はDeep Learningだけでは解決不能で企業のルール等も鑑みながらロジックを組み立てており、Deep Learningは異常検知の機能の一部分でしかないことを強調していた事が非常に印象的でした。

RAY: A CLUSTER COMPUTING ENGINE FOR REINFORCEMENT LEARNING APPLICATIONS

概要

UC Berkeleyの学生たちによるSpark をベースとした強化学習のフレームワーク(Ray)についてのプレゼン

機械学習が成熟するにつれて、標準的な教師あり学習は学習データの量と正確性という観点から十分な手段ではなくなりました。

昨今の機会学習アプリケーションでは、静的に学習したモデルから単一の予測を作成して提供する代わりに、一連のアクションを実行した結果を再評価する事で、動的環境の変化に対応する必要性が高まっています。強化学習という手段はこの要求に対して良い成果を出すことが知られています。 ただし、これらのアプリケーションは非常に厳しい計算要件を必要とし、動的なグラフ計算もサポートする必要があります。彼らは、この要件に対してスケールアウトが容易であるSparkをコンピューティングエンジンとして選択しており、そのアーキテクチャー及び一部実装方法についてがプレゼンの内容となっています。

感想

強化学習は違ったパラメータでシミュレーションを行いそれを評価してルールを更新し、再度違ったパラメータでシミュレーションを行い再評価して。。。という事をするのですが、彼らはそれを並列化して大量にシミュレーションを行いたいと、、、これこそ、Sparkの出番!!という感じで、単純にああなるほどね。と納得しました。

また講演では、人の走るという行為を強化学習でシミュレーションするサンプルアプリケーションのデモ動画が流されていて、最初の方は一歩進むのもままならない状態のものが、1歩・2歩と進むようになり、最終的には(不格好ながらも)ずっと走り続けるようになる結果を見て、単純におもしろいなと思いました。

GoogleのAlphaGoが強化学習とDeep Learningを利用して強くなったことはあまりに有名ですが、エンタープライズの世界でも(Sparkをベースとした!?)強化学習によりブレークスルーが起きるか要注目ですね。

BIG DATA AT AUDI: ROOT CAUSE ANALYSIS IN AN AUTOMOTIVE PAINT SHOP USING MLLIB

f:id:brains_aoki:20170608135540p:plain:w160f:id:brains_aoki:20170608135631p:plain:w160 f:id:brains_aoki:20170608135714p:plain:w160

概要

アウディの自動車塗装において取得できる、約2500のセンサーデータを活用した話。塗装工程は非常に複雑かつ、温度や湿度等の影響をうけるため、どの工程でどのような失敗が起きるのかセンサーデータにSparkのDataFrame APIやMLlibを使って原因を突き止めることを試みたとのことです。

感想

個人的に興味深かったのは、時刻調整や補正に関する話です。自動車が塗装されるタイミングにおけるセンサーデータの時刻調整の話や、ある時間間隔におけるデータの代表値を平均か回帰で補正するのかなど、弊社も似たようなことをやっているなと感じました。こういった親近感のような感想を得られたことは、ある意味収穫ではありました。

終わりに

Keynoteでもありましたが、これからSparkはDeep Learningのサポート・ストリーミングのマイクロバッチ廃止など、更に進化していきます。個人的にはDeep Learningの機能は早く使ってみたいですし、とても楽しみにしています。 今後もまだまだ、Sparkの進化から目が離せませんね!

【レポート】Spark Summit 2017 開幕!!!

Spark Summit 2017 San Francisco

f:id:brains_aoki:20170607004557j:plain

こんにちは、データアナリストの青木とエンジニアの樋口です。

6月5日から合計3日間アメリカのサンフランシスコでSpark Summit2017が開催されています。

spark-summit.org

Spark Summit2017では、3,000人以上のエンジニア、データサイエンティスト、研究者やビジネスプロフェッショナルが参加しており、データサイエンスやエンタープライズ、マシンラーニングなどの分野で、170を超えるセッションがあります。

ブレインズテクノロジーではデータ分析・機械学習を事業のコアとしている事もありSparkの動向には常に注目しています。

その中で、CTOから「ちょっと、きみたち行ってきたら?」というありがたい鶴の一声を頂いたのではるばるアメリカへ来ています!

セッションの内容等、順次アップして行く予定ですが、まずは雰囲気だけ!!

f:id:brains_aoki:20170607004445j:plain f:id:brains_aoki:20170607004526j:plain f:id:brains_aoki:20170607010319j:plain

(P.S サンフランシスコ、最高!)

f:id:brains_aoki:20170607010353j:plain

【レポート】ケイ・オプティコム様 IBMユーザ論文「AIを利用したサイレント故障検知の取組み」で金賞受賞!

データ分析サービス・プロダクト担当の藤原です。

弊社のリアルタイム大規模データ分析基盤「Impulse」を導入いただきましたケイ・オプティコム様が、ネットワークインフラ運用において機械学習を活用してサイレント故障の検知を実現した取組みについて、IBMが主催するIBMユーザー研究会の論文を執筆され、「金賞」を受賞されました!
http://www.uken.or.jp/symp/symp55/program/paperlist.shtml

関西エリアを中心として大規模なネットワークインフラの安定稼働のため、ネットワークのサイレント故障の迅速な検知を実現する手段として、機械学習に着目され、弊社Impulseを活用して分析基盤を整備された取組みの詳細がまとめられた内容です。

5月18〜19日には「IBMユーザー・シンポジウム」として、約1,000人の来場者が訪れる大規模なイベントが京都で開催されました。
www.uken.or.jp

機械学習の活用に関する具体的な取組み事例ということもあり、ケイ・オプティコム様のプレゼン会場は超満員。
聴講されていた方々も非常に真剣な様子で、その内容に聞き入っておられました。

今回の金賞受賞は、ケイ・オプティコム様が「AI・機械学習」といった技術を、いち早く「システムインフラ・ネットワーク運用」という分野で、実際の現場に適用されたという先進的な取組みであったという点が高い評価に繋がっているではないかと思います。
そうした貴重な取組みに対して、技術的な面で貢献できたことを弊社としても非常に嬉しく感じております。

今後もケイ・オプティコム様を始めとした多くの導入企業様の事業に貢献できるよう、Impulseの更なる進化に取り組んでいきたいと思います。

f:id:fujiwarakazunari:20170531135705j:plain
※シンポジウムでの表彰後の様子:株式会社ケイ・オプティコム 谷岡様(右)と赤井様(左)

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の「これから」にご期待ください!