ビッグデータシグナル解析：シグナルデータを保存して問い合わせるためのより良い方法

Question

私はHadoop/Sparkでいくつかのシグナル分析を行い、プロセス全体をどのように構築するかについて助言が必要です。

信号が、今私たちはSqoopで読まれ、に似てスキーマを使用して、HDFS上のファイルに変換されますことを、データベースに格納されています。

<Measure ID> <Source ID> <Measure timestamp> <Signal values> 

信号値は浮動小数点で作られた単なる文字列です

コンマで区切られた数字。我々は、対話型/バッチクエリを記述したいと思い

00S001 2015/04/22T10:00:00.000Z 0.0,1.0,200.0,30.0 ... 100.0 
000124 S001 2015/04/22T10:05:23.245Z 0.0,4.0,250.0,35.0 ... 10.0 
... 
000126 S003 2015/04/22T16:00:00.034Z 0.0,0.0,200.0,00.0 ... 600.0 

：

は、信号の上に集約関数を適用1000.0の上にピークを持っていた信号を選択するには

SELECT * 
FROM SIGNALS 
WHERE MAX(VALUES) > 1000.0 

値。

は1500.0を超え、少なくとも一つの信号を有するソースを選択する

SELECT SOURCEID, MAX(VALUES) FROM SIGNALS GROUP BY SOURCEID HAVING MAX(MAX(VALUES)) > 1500.0 

集合にわたって集約を適用します。

は5.0 kHzで濾過された後に100.0上に少なくとも値を有する信号を選択するためにサンプル

SELECT * 
FROM SIGNALS 
WHERE MAX(LOW_BAND_FILTER("5.0 KHz", VALUES)) > 100.0) 

介してユーザ定義関数を適用します。

我々はためにいくつかの助けを必要とする：HDFS上の信号データを書き込むための正しいファイル形式を見つける

1. 。私はApache Parquetを考えました。どのようにデータを構造化しますか？
2. データ分析の適切なアプローチを理解してください。さまざまなデータセット（Sparkでデータを処理し、HDFSで結果を永続化するなど）を作成するか、元のデータセットからクエリ時にすべてを実行する方がよいでしょうか？
3. は、私が書いたようなクエリを作成するのに最適なツールですか？私たちはCloudera Enterprise Hadoopを実行しているので、Impalaも使用できます。
4. 元のデータセットとは異なる派生データセットを作成する場合、データの系列をどのように追跡することができますか？つまり、データが元のバージョンからどのように生成されたかを知ることができますか？

ありがとうございました！

Answer 1

1）円柱形式の寄木細工はOLAPに適しています。 Parquetのスパークサポートは、プロダクションで使用するのに十分なほど成熟しています。私は、データ構造（単純化）を以下に信号値を表す文字列を解析することをお勧め：あなたには、いくつかの種類が必要な場合は

def maxV(arr: mutable.WrappedArray[Double]) = arr.max 
sqlContext.udf.register("maxVal", maxV _) 
df.registerTempTable("table") 

sqlContext.sql("select * from table where maxVal(signals) > 2.1").show() 
+---+---------------+ 
| id|  signals| 
+---+---------------+ 
| 2|  [3.0, 5.0]| 
| 2|[1.5, 7.0, 8.0]| 
+---+---------------+ 

sqlContext.sql("select id, max(maxVal(signals)) as maxSignal from table group by id having maxSignal > 1.5").show() 
+---+---------+ 
| id|maxSignal| 
+---+---------+ 
| 1|  2.0| 
| 2|  8.0| 
+---+---------+ 

または、：

case class Data(id: Long, signals: Array[Double]) 
val df = sqlContext.createDataFrame(Seq(Data(1L, Array(1.0, 1.0, 2.0)), Data(2L, Array(3.0, 5.0)), Data(2L, Array(1.5, 7.0, 8.0)))) 

doubleの配列を維持することは、このようなUDFを定義して使用することができます-safety、ScalaのDSLを使用して：

import org.apache.spark.sql.functions._ 
val maxVal = udf(maxV _) 
df.select("*").where(maxVal($"signals") > 2.1).show() 
df.select($"id", maxVal($"signals") as "maxSignal").groupBy($"id").agg(max($"maxSignal")).where(max($"maxSignal") > 2.1).show() 
+---+--------------+ 
| id|max(maxSignal)| 
+---+--------------+ 
| 2|   8.0| 
+---+--------------+ 

2）それは異なります。あなたのデータのサイズは、合理的なレイテンシでクエリ時間ですべての処理を行うことができますならば - それのために行きます。このアプローチから始めて、遅い/一般的なクエリのための最適化された構造を後で構築することができます。

3）ハイブは遅く、ImpalaとSpark SQLによって時代遅れです。 HDFS/Hive、Sparkに格納されているすべてのデータのレイテンシは大きいが、結合は信頼性が高く、より多くのデータソースをサポートし、豊富な非SQL処理を備えている場合、Impalaは結合のないクエリに適しています。 （MLlibやGraphXのような）

4）生データ（マスターデータセット）を重複しないように保存します（時間ベースのパーティションを使用します）。新しいデータがパーティションに到着し、すでにダウンストリームデータセットが生成されている場合は、このパーティションのパイプラインを再起動します。

ホープこれは私がVitaliyのアプローチは、あらゆる面で非常に良好であると信じて、

Answer 2

まず役立ちます。 （そして、私はスパークのためにすべてです）

もう一つのアプローチを提案したいと思います。理由は以下のとおりです。

1. 我々は（+私たちはCDHを持っている）
2. 既にデータが必要「の分析」とデータのない、非常に「処理」にある
3. を構造化された対話型のクエリを実行したいです。 （a）は、データが構造化されている場合は、スパークはやり過ぎかもしれない、我々はより速く、SQLクエリを形成することができ、（b）は、私たちはここで、クエリ

を実行するたびにステップでプログラムを書くことはしたくありません私は行きたい：HDFSにsqoopを使用して

1. 飲み込んだ場合：[必要に応じて]
2. --as-parquetfileを使用し、あなたが望むように外部インパラテーブルまたは内部テーブルを作成します。ファイルを寄木細工のファイルとして転送していない場合は、この手順でファイルを転送できます。私たちのグループ分けがその列で起こるので、好ましくはソースIDによってパーティション化します。

基本的に、データが転送されたら、好ましくは寄木張りの形式でImpalaテーブルを作成し、グループ化に使用する列で分割します。読み込み後に統計情報を計算して、Impalaがより速く実行できるようにしてください。

データの移動： - 私たちは別のファイル を作成し、その結果のうち、フィードを生成する必要がある場合 - 別のシステムは、既存のデータを更新するために起こっている場合は、creating->積載しながら、別の場所にデータを移動テーブル - クエリと分析についてのみで、レポートを取得する場合（i。外部テーブルで十分です）、データを不必要に移動する必要はありません - 同じデータの上に外部ハイブテーブルを作成できます。長期実行クエリを実行する必要がある場合は、Hiveを使用できます。しかし、インタラクティブなクエリの場合はノー・ノーです。ヒントから派生したテーブルを作成し、Impalaを使用する場合は、ハイブ生成テーブルでインパラクエリを実行する前に 'メタデータを無効にする'を実行することを忘れないでください。

リニージ - ここでは具体的には、link on Impala lineage using Cloudera Navigator