ツェッペリンにリアルタイムグラフをプロットする方法は？

Question

私は、リアルタイムグラフをプロットするためにゼップリンを使用しようとしています。私は毎分つぶやきについて感情分析をしています。私は静的にクエリを行い、グラフをプロットすることができます。しかし、私はこれが動的に行われることを望みます。私はzeppelinには新しく、angularJSについて多くの知識を持っていません。この問題に対する正しいアプローチは何でしょうか？

val final_score=uni_join.map{case((year,month,day,hour,minutes),(tweet_count,sentiment))=>(year, month, day, hour, minutes(sentiment/tweet_count).ceil)} 


final_score.saveToCassandra("twitter", "score",writeConf = WriteConf(ttl = TTLOption.constant(1000))) 

final_score.foreachRDD(score => { 
val rowRDD =score.map{case(year,month,day,hour,minutes,sentiment) =>(year,month,day,hour,minutes,sentiment) } 
    val tempDF = sqlContext.createDataFrame(rowRDD) 

    z.angularBindGlobal("stream", parsed) //to bind parsed to stream. 
    tempDF.registerTempTable("realTimeTable") 
}) 

グラフを取得することができます。しかし、感情スコアに同期するために毎分グラフを動的に更新したいと思います。 ありがとうございます。次のように [更新]ツェッペリンのノートブックの角部は、次のとおりです。

%angular 
<div id="graph" style="height: 100%; width: 100%"> 
<canvas id="myChart" width="400" height="400"></canvas> 
<div id="legendDiv"></div> 
</div> 
<script> 
function initMap() { 

var colorList = ["#fde577", "#ff6c40", "#c72a40", "#520833", "#a88399"] 


var el = angular.element($('#stream')); 

console.log("El is "+el) //returns el as object 

angular.element(el).ready(function() { 
    console.log('Hello') 
    window.locationWatcher =el.$scope.$watch('stream', function(new, old){ 
console.log('changed');}, true)}) 
</script> 

しかし、このコードを実行するには、次のエラーを返し続けます。私が使用しています

vendor.js:29 jQuery.Deferred exception: Cannot read property '$watch' of undefined TypeError: Cannot read property '$watch' of undefined 

スパークバージョンは1.6 あり、ツェッペリンは、バージョン0.6.3のサポートSpark Structured Streaming以来の0.6

Answer 1

spark-highchartsです。

集計後のstructuredDataFrameについては、以下のコードを1つのツェッペリンの段落に入れてください。 OutputModeは、appendまたはcompleteのいずれかで、structureDataFrameの集計方法に依存します。

import com.knockdata.spark.highcharts._ 
import com.knockdata.spark.highcharts.model._ 

val query = highcharts(
    structuredDataFrame.seriesCol("country") 
    .series("x" -> "year", "y" -> "stockpile") 
    .orderBy(col("year")), z, "append") 

そして次の段落のコード。この段落のチャートは、新しいデータがstructureDataFrameに来ると更新されます。

StreamingChart(z) 

次のコードを実行すると、グラフの更新が停止します。

ここでは、generateDataFrameの例を示します。

spark.conf.set("spark.sql.streaming.checkpointLocation","/usr/zeppelin/checkpoint") 

case class NuclearStockpile(country: String, stockpile: Int, year: Int) 

val USA = Seq(0, 0, 0, 0, 0, 6, 11, 32, 110, 235, 369, 640, 
    1005, 1436, 2063, 3057, 4618, 6444, 9822, 15468, 20434, 24126, 
    27387, 29459, 31056, 31982, 32040, 31233, 29224, 27342, 26662, 
    26956, 27912, 28999, 28965, 27826, 25579, 25722, 24826, 24605, 
    24304, 23464, 23708, 24099, 24357, 24237, 24401, 24344, 23586, 
    22380, 21004, 17287, 14747, 13076, 12555, 12144, 11009, 10950, 
    10871, 10824, 10577, 10527, 10475, 10421, 10358, 10295, 10104). 
    zip(1940 to 2006).map(p => NuclearStockpile("USA", p._1, p._2)) 

val USSR = Seq(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 
    5, 25, 50, 120, 150, 200, 426, 660, 869, 1060, 1605, 2471, 3322, 
    4238, 5221, 6129, 7089, 8339, 9399, 10538, 11643, 13092, 14478, 
    15915, 17385, 19055, 21205, 23044, 25393, 27935, 30062, 32049, 
    33952, 35804, 37431, 39197, 45000, 43000, 41000, 39000, 37000, 
    35000, 33000, 31000, 29000, 27000, 25000, 24000, 23000, 22000, 
    21000, 20000, 19000, 18000, 18000, 17000, 16000). 
    zip(1940 to 2006).map(p => NuclearStockpile("USSR/Russia", p._1, p._2)) 

input.addData(USA.take(30) ++ USSR.take(30)) 
val structureDataFrame = input.toDF 

また、次のコードをシミュレーションしてグラフを更新することもできます。次のコードを実行すると、グラフが更新されます。

input.addData(USA.drop(30) ++ USSR.drop(30)) 

NOTE: The example using Zeppelin 0.6.2 and Spark 2.0

NOTE: Please check the highcharts license for commercial usage