# html 그리는 것을 파이썬 언어로 할 수 있음

import dash
import dash_core_components as dcc
import dash_html_components as html

external_stylesheets = ['https://codepen.io/chriddyp/pen/bWLwgP.css']

app = dash.Dash(__name__, external_stylesheets=external_stylesheets)
# dash app 초기화

# dash app에 무엇을 띄울 것인가에 대한 정의
# app.layout은 레이아웃을 지정
# html 그리는 것을 html, 자바스크립트가 아니라 아래와 같이 파이썬 코드로 작성하는 것이 dash의 특징이다.
app.layout = html.Div(children=[
    html.H1(children='Hello Dash'),

    html.Div(children='''
        Dash: A web application framework for Python.
    '''),

    dcc.Graph(
        id='example-graph',
        figure={
            'data': [
                {'x': [1, 2, 3], 'y': [4, 1, 2], 'type': 'bar', 'name': 'SF'},
                {'x': [1, 2, 3], 'y': [2, 4, 5], 'type': 'bar', 'name': u'Montréal'},
            ],
            'layout': {
                'title': 'Dash Data Visualization'
            }
        }
    )
])

if __name__ == '__main__':
    app.run_server(debug=True)
    # 실행을하면 서버가 뜸

export CLASSPATH=`pwd`/target/kafka-connect-twitter-0.1-jar-with-dependencies.jar

export CLASSPATH=`pwd`/target/kafka-connect-twitter-0.1-jar-with-dependencies.jar

# -*- coding: utf-8 -*-
from flask import Flask, jsonify, abort, make_response, request, url_for
import ast

import dash
import dash_core_components as dcc
import dash_html_components as html
import plotly.graph_objs as go
from dash.dependencies import Input, Output

labels = ['Waiting']
values = [0]

server = Flask(__name__)

@server.route('/test', methods=['GET'])
def test_api_request():
    print('IM IN ROUTE /TEST')
    return 'success'

@server.route('/update_data', methods=['POST'])
def update_data():
    print('IM IN ROUTE /UPDATE_DATA')
    
    global labels, values
    if not request.form or 'tag_count' not in request.form:
        return "error",400
    print(request)
    print(request.form)
    labels = ast.literal_eval(request.form['tag'])
    labels.reverse()
    values = ast.literal_eval(request.form['tag_count'])
    values.reverse()
    print("labels received: " + str(labels))
    print("data received: " + str(values))
    return "success",201
# labels, values로 초기화해주고, 플라스크로 만들어놓은 api에 의해서 value랑 label을
# 입력을 받아서 넣어주는 형태가 된다.
# @server.route('/update_data', methods=['POST'])
# def update_data():
# 위에 두줄처럼 하면 update_data라는 api가 만들어진다.
# request가 들어왔을때 처리방법을 구현해놓았다.
# 리퀘스트가 들어오면 검사같은걸하고 디버깅을 위한 프린팅을 한다음에
# ast.literal_eval(request.form['tag']) 또는 ast.literal_eval(request.form['tag_count'])해서
# hashtag_ranking.py에서 보내준 post의 데이터를 받아서 labels랑 values에 넣어주게 된다.

app = dash.Dash(__name__, server=server)
# 대시보드를 여는 것으로 server=server는 
# 위에서 수동으로 코드로 작성한 flask를 임의로 초기화를 시켜서 서버를 넣어준다.

app.layout = html.Div(children=[
    html.H1('Twitter stream DashBoard'),

    dcc.Graph(
        id='live-update-graph'
    ),
    dcc.Interval(
        id='interval-component',
        interval=1*1000, # in milliseconds
        n_intervals=0
    )
])
# 그 다음에 대시보드 layout에 하나의 그래프를 넣어주는데 
# 그 그래프는 라이브 업데이트 그래프라고해서 인터벌을 넣어주면
# 얘를 1초마다 업데이트하는 그래프이다.
# 그래서 dcc.Graph를 넣어주고 그 밑에는 눈에 보이지는 않지만 
# 인터벌 업데이트를 담당하는 컴포넌트를 넣어준다.

# 인터널 컴포넌트가 인풋아웃풋으로해서 인터널컴포넌트에 업데이트가 되었을때
# 라이브업데이트그래프를 업데이트하라는 의미
# 여기에다 figure에 데이터를 넣어서 피큐어를 리턴을 해주면 이게 그래프에 들어가게된다.
# 들어오는 데이터는 벨류와 레이블로 넣을것인데 가로방향으로 그릴것이다.
# x에 value, y에 label을 넣을 것이다.
# 벨류와 레이블은 위에 update_data 함수있는곳에서 초기화 한다.
# 그리고 이 api는 update_data라는 api를 호스트로 호출하면 그 데이터를 뽑아서
# label이랑 value로 넣어주는 api이다.
@app.callback(Output('live-update-graph', 'figure'),
              [Input('interval-component', 'n_intervals')])
def update_graph_live(n):

    figure={
        'data': [
            go.Bar(x=values, 
                    y=labels,
                    orientation='h',
                    width=0.2)
        ],
        'layout': {
            'title': 'Dash Data Visualization 13'
        }
    }
    return figure

if __name__ == '__main__':
    app.run_server(debug=True, dev_tools_hot_reload=True)
# dev_tools_hot_reload=True하면 코드 업데이트하면 바로 업데이트가 자동으로 된다.

import requests

import findspark
findspark.init()
import pyspark

from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()

from pyspark.sql.types import *
from pyspark.sql.functions import *

spark.sparkContext.setLogLevel("ERROR")

kafka_df = spark.readStream.format("kafka").option("kafka.bootstrap.servers", "localhost:9092").option("subscribe", "test").load()
# 스파크 데이터프레임에 spark.readStream.format("kafka") 이런식으로 명령어를 입력하면 카프카를 읽어드릴 수 있다.
# 그 다음에 .option("kafka.bootstrap.servers", "localhost:9092").option("subscribe", "test") 명령어를 보면 콘솔컨수머를 띄울때랑 똑같은
# 옵션들이 들어간다. 부트스트랩 서버에 로컬호스트 9092, 그다음에 test 토픽에다 올려준다. 이런뜻이고 이렇게 한다음에 로드하면
# 데이터 프레임이 만들어 지는 것이다.
kafka_df_string = kafka_df.select(col("key").cast("STRING").alias("key"),col("value").cast("STRING").alias("value"))
# 그 다음에 데이터가 키벨류 형태로 들어오는데 

kafka_df_string_2 = kafka_df_string.select(col("value"))
# 그거를 키는 무시고 벨류만 가져오게 select을 하고

kafka_df_tags = kafka_df_string_2.select(explode(split('value', ' ')).alias('tag')).filter(col('tag').startswith('#'))
# #(해시테그)으로 시작하는 것만 select함
kafka_df_tag_count = kafka_df_tags.groupBy('tag').count().withColumnRenamed('count', 'tag_count').orderBy(col('count').desc())
# #(해시테그) 개수가 많은것만 뽑아진다.
# 결론적으로 위에 두줄의 코드는 들어오는 데이터에 대해 #(해시테그)로 시작하는 것을 wordcount해 갯수가 많은 단어를 순서대로 보여주게 된다.

output = kafka_df_tag_count.writeStream.outputMode("complete").format("console").option("truncate", "false").trigger(processingTime="3 seconds").start()
# kafka_df_tag_count 변수를 writeStream.outputMode에서 콘솔에다가 업데이트 모드로 뿌려주는 것을 의미한다. 
# 그리고 3초 주기로 배치테이블을 뿌려준다는 것이다.

def send_df_to_dashboard(df, id):
    tag = [str(t.tag) for t in df.select("tag").take(10)]
    tag_count = [str(t.tag_count) for t in df.select("tag_count").take(10)]
    url = 'http://localhost:8050/update_data'
    request_data = {'tag': str(tag), 'tag_count': str(tag_count)}
    print('update dashboard')
    response = requests.post(url, data=request_data)
# 데이터프레임의 값을 가져와서 df.select("tag"), df.select("tag_count") 두개를 뽑아준다.
# 얘가 로우라는 클래스로 나오는데 얘를 뽑으려면 .take()를 해줘야한다.
# 이렇게 뽑은 형태를 딕셔너리 형태로 만들어서(request_data)
# localhost:8050/update_data라는 곳에 
# requests라는 파이썬의 기본 모듈을 이용해서 post를 해준다.
# post한건 어디서 받냐면 dash 띄웠던 곳 app3.py에 넘긴다.

kafka_df_tag_count.writeStream.outputMode("complete").foreachBatch(send_df_to_dashboard).trigger(processingTime="3 seconds").start()
# kafka_df_tag_count 변수를 writeStream.outputMode에서 foreachBatch라는 api를 이용해서
# batch를 할때마다 send_df_to_dashboard 함수를 실행하게 한다.
# 그래서 dash에다가 업데이트 모드로 3초 주기로 뿌려주는 것을 의미한다. 

output.awaitTermination()

./spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.0 hashtag_ranking.py

./pyspark --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.0

from pyspark.sql.types import *
from pyspark.sql.functions import *

kafka_df = spark.readStream.format("kafka").option("kafka.bootstrap.servers", "localhost:9092").option("subscribe", "test").load()
kafka_df_string = kafka_df.select(col("key").cast("STRING").alias("key"),col("value").cast("STRING").alias("value"))

kafka_df_string_2 = kafka_df_string.select(col("value"))

output = kafka_df_string_2.writeStream.outputMode("update").format("console").option("truncate", "false").trigger(processingTime="5 seconds").start()

from pyspark.sql.types import *
from pyspark.sql.functions import *

kafka_df = spark.readStream.format("kafka").option("kafka.bootstrap.servers", "localhost:9092").option("subscribe", "test").load()
kafka_df_string = kafka_df.select(col("key").cast("STRING").alias("key"),col("value").cast("STRING").alias("value"))

kafka_df_string_2 = kafka_df_string.select(col("value"))

kafka_df_tags = kafka_df_string_2.select(explode(split('value', ' ')).alias('name')).filter(col('name').startswith('#'))
kafka_df_tag_count = kafka_df_tags.groupBy('name').count().orderBy(col('count').desc())

output2 = kafka_df_tag_count.writeStream.outputMode("complete").format("console").option("truncate", "false").trigger(processingTime="5 seconds").start()