ปัจจุบันข้อมูลมีการเติบโตและสะสมเร็วกว่า แต่ก่อน ปัจจุบันประมาณ 90% ของข้อมูลที่สร้างขึ้นในโลกของเราถูกสร้างขึ้นในช่วงสองปีที่ผ่านมา เนื่องจากอัตราการเติบโตนี้แพลตฟอร์มต่างๆ ข้อมูลใหญ่ พวกเขาต้องใช้วิธีแก้ปัญหาที่รุนแรงเพื่อให้สามารถรักษาข้อมูลจำนวนมากเช่นนี้ได้
ผลกระทบของสีต่อผู้คน
แหล่งข้อมูลที่สำคัญที่สุดแหล่งหนึ่งในปัจจุบันคือโซเชียลมีเดีย ให้ฉันแสดงตัวอย่างชีวิตจริง: การจัดการวิเคราะห์และดึงข้อมูลจากข้อมูลโซเชียลมีเดียแบบเรียลไทม์โดยใช้หนึ่งในโซลูชันด้านสิ่งแวดล้อมใน ข้อมูลใหญ่ สิ่งที่สำคัญที่สุด - Apache Spark และ Python
ในบทความนี้ฉันจะแสดงวิธีสร้างแอปพลิเคชั่นง่ายๆที่อ่านฟีดออนไลน์ของ Twitter โดยใช้ Python จากนั้นประมวลผลทวีตโดยใช้ Apache Spark Streaming เพื่อระบุแฮชแท็กและสุดท้ายส่งคืนแฮชแท็กที่กำลังมาแรงที่สำคัญที่สุดและแสดงข้อมูลนี้บนแดชบอร์ดแบบเรียลไทม์
การสร้างข้อมูลรับรองของคุณเองสำหรับ Twitter API
ในการรับทวีตจาก Twitter คุณต้องลงทะเบียนที่ TwitterApps โดยคลิกที่ 'สร้างแอปพลิเคชันใหม่' และหลังจากกรอกแบบฟอร์มด้านล่างแล้วให้คลิกที่ 'สร้างแอปพลิเคชัน Twitter ของคุณ'
อย่างที่สองไปที่แอปพลิเคชันที่คุณสร้างขึ้นใหม่และเปิดหน้าต่าง 'ตัวระบุและคีย์การเข้าถึง' จากนั้นคลิกที่ 'สร้างตัวระบุการเข้าถึงของฉัน'
รหัสเข้าสู่ระบบใหม่ของคุณจะปรากฏดังที่แสดงด้านล่าง
และตอนนี้คุณก็พร้อมสำหรับขั้นตอนต่อไป
สร้างไคลเอนต์ HTTP Twitter
ในขั้นตอนนี้ฉันจะแสดงวิธีสร้างไคลเอนต์อย่างง่ายที่จะดึงทวีตจาก Twitter API โดยใช้ Python จากนั้นส่งต่อไปยังอินสแตนซ์ Spark Streaming . มันควรจะง่ายต่อการปฏิบัติสำหรับใด ๆ นักพัฒนา python มืออาชีพ.
ขั้นแรกเราจะสร้างไฟล์ชื่อ twitter_app.py จากนั้นเราจะเพิ่มโค้ดเข้าด้วยกันดังที่แสดงด้านล่าง
หมายเลขบัตรเครดิตที่ใช้กับ cvv 2017
นำเข้าไลบรารีที่เราจะใช้ดังที่แสดงด้านล่าง:
import socket import sys import requests import requests_oauthlib import json และเพิ่มตัวแปรที่จะใช้ใน OAuth เพื่อเชื่อมต่อกับ Twitter ดังที่แสดงด้านล่าง:
# Reemplaza los valores de abajo con los tuyos ACCESS_TOKEN = 'YOUR_ACCESS_TOKEN' ACCESS_SECRET = 'YOUR_ACCESS_SECRET' CONSUMER_KEY = 'YOUR_CONSUMER_KEY' CONSUMER_SECRET = 'YOUR_CONSUMER_SECRET' my_auth = requests_oauthlib.OAuth1(CONSUMER_KEY, CONSUMER_SECRET,ACCESS_TOKEN, ACCESS_SECRET) ตอนนี้เรามาสร้างฟังก์ชันใหม่ชื่อ get_tweets ซึ่งจะเรียก URL API ของ Twitter และส่งคืนการตอบสนองสำหรับสตริงทวีต
def get_tweets(): url = 'https://stream.twitter.com/1.1/statuses/filter.json' query_data = [('language', 'en'), ('locations', '-130,-20,100,50'),('track','#')] query_url = url + '?' + '&'.join([str(t[0]) + '=' + str(t[1]) for t in query_data]) response = requests.get(query_url, auth=my_auth, stream=True) print(query_url, response) return response จากนั้นคุณสร้างฟังก์ชันที่รับการตอบสนองจากมุมมองด้านบนและแยกข้อความของทวีตจากออบเจ็กต์ JSON ของทวีตแบบเต็ม หลังจากนี้ให้ส่งทวีตแต่ละรายการไปยังอินสแตนซ์ Spark Streaming (จะกล่าวถึงในภายหลัง) ผ่านการเชื่อมต่อ TCP
def send_tweets_to_spark(http_resp, tcp_connection): for line in http_resp.iter_lines(): try: full_tweet = json.loads(line) tweet_text = full_tweet['text'] print('Tweet Text: ' + tweet_text) print ('------------------------------------------') tcp_connection.send(tweet_text + '
') except: e = sys.exc_info()[0] print('Error: %s' % e) ตอนนี้เราจะทำส่วนหลัก สิ่งนี้จะทำให้แอปพลิเคชันโฮสต์การเชื่อมต่อ เบ้า ซึ่งจะเชื่อมต่อในภายหลัง จุดประกาย . เราจะกำหนดค่า IP ที่นี่ให้เป็น localhost เนื่องจากทุกอย่างจะทำงานบนเครื่องเดียวกันและบนพอร์ต 9009. จากนั้นเราจะเรียกวิธี get_tweets ซึ่งเราได้ทำข้างต้นเพื่อรับทวีตจาก Twitter และส่งการตอบกลับของคุณด้วยการเชื่อมต่อ เบ้า ก send_tweets_to_spark เพื่อส่งทวีตไปยัง Spark
TCP_IP = 'localhost' TCP_PORT = 9009 conn = None s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.bind((TCP_IP, TCP_PORT)) s.listen(1) print('Waiting for TCP connection...') conn, addr = s.accept() print('Connected... Starting getting tweets.') resp = get_tweets() send_tweets_to_spark(resp, conn) การติดตั้งแอปพลิเคชันสตรีมมิ่ง Apache Spark ของเรา
มาสร้างแอปพลิเคชันของเรา Spark Streaming ซึ่งจะทำการประมวลผลแบบเรียลไทม์สำหรับทวีตที่เข้ามาแยกแฮชแท็กจากพวกเขาและคำนวณจำนวนแฮชแท็กที่ถูกกล่าวถึง
ขั้นแรกเราต้องสร้างอินสแตนซ์ บริบทจุดประกาย sc จากนั้นเราจะสร้าง บริบทการสตรีม ssc ของ sc ด้วยช่วงเวลาสองวินาทีที่จะทำการเปลี่ยนแปลงในการส่งสัญญาณทั้งหมดที่ได้รับทุกๆสองวินาที โปรดทราบว่าเราตั้งค่าระดับการบันทึกเป็น ERROR เพื่อให้สามารถปิดใช้งานบันทึกส่วนใหญ่ที่คุณเขียน จุดประกาย .
เรากำหนดจุดตรวจที่นี่เพื่อให้สามารถตรวจสอบ RDD ได้เป็นระยะ สิ่งนี้จำเป็นต้องใช้ในแอปพลิเคชันของเราเนื่องจากเราจะใช้การแปลงไฟสถานะ (จะกล่าวถึงต่อไปในหัวข้อเดียวกัน)
จากนั้นเรากำหนด DStream dataStream หลักของเราซึ่งจะเชื่อมต่อเซิร์ฟเวอร์ เบ้า ที่เราสร้างไว้ก่อนหน้านี้ในพอร์ต 9009 และจะอ่านทวีตจากพอร์ตนั้น ทุกบันทึกใน DStream จะเป็นทวีต
from pyspark import SparkConf,SparkContext from pyspark.streaming import StreamingContext from pyspark.sql import Row,SQLContext import sys import requests # crea una configuración spark conf = SparkConf() conf.setAppName('TwitterStreamApp') # crea un contexto spark con la configuración anterior sc = SparkContext(conf=conf) sc.setLogLevel('ERROR') # crea el Contexto Streaming desde el contexto spark visto arriba con intervalo de 2 segundos ssc = StreamingContext(sc, 2) # establece un punto de control para permitir la recuperación de RDD ssc.checkpoint('checkpoint_TwitterApp') # lee data del puerto 9009 dataStream = ssc.socketTextStream('localhost',9009) ตอนนี้เราจะกำหนดตรรกะการเปลี่ยนแปลงของเรา ก่อนอื่นเราจะแบ่งทวีตทั้งหมดออกเป็นคำ ๆ และใส่ลงในคำ RDD จากนั้นเรากรองเฉพาะแฮชแท็กของคำทั้งหมดและพล็อตไว้ข้างๆ (hashtag, 1) และใส่ไว้ในแฮชแท็ก RDD
จากนั้นเราต้องคำนวณว่าแฮชแท็กถูกพูดถึงกี่ครั้ง เราสามารถทำได้โดยใช้ฟังก์ชัน reduceByKey ฟังก์ชันนี้จะคำนวณจำนวนครั้งที่แต่ละกลุ่มพูดถึงแฮชแท็กนั่นคือจะรีเซ็ตบัญชีในแต่ละกลุ่ม
ในกรณีของเราเราจำเป็นต้องคำนวณจำนวนในทุกกลุ่มดังนั้นเราจะใช้ฟังก์ชันอื่นที่เรียกว่า updateStateByKey เนื่องจากฟังก์ชันนี้ช่วยให้คุณสามารถรักษาสถานะ RDD ในขณะที่อัปเดตด้วยข้อมูลใหม่ แบบฟอร์มนี้เรียกว่า Stateful Transformation
โปรดทราบว่าในการใช้ updateStateByKey คุณต้องกำหนดค่าจุดตรวจและสิ่งที่ทำในขั้นตอนก่อนหน้านี้
# divide cada Tweet en palabras words = dataStream.flatMap(lambda line: line.split(' ')) # filtra las palabras para obtener solo hashtags, luego mapea cada hashtag para que sea un par de (hashtag,1) hashtags = words.filter(lambda w: '#' in w).map(lambda x: (x, 1)) # agrega la cuenta de cada hashtag a su última cuenta tags_totals = hashtags.updateStateByKey(aggregate_tags_count) # procesa cada RDD generado en cada intervalo tags_totals.foreachRDD(process_rdd) # comienza la computación de streaming ssc.start() # espera que la transmisión termine ssc.awaitTermination() updateStateByKey รับฟังก์ชันเป็นพารามิเตอร์ที่เรียกว่าฟังก์ชัน update สิ่งนี้ดำเนินการกับแต่ละรายการใน RDD และดำเนินการตามตรรกะที่ต้องการ
ในกรณีของเราเราได้สร้างฟังก์ชันการอัปเดตที่ชื่อว่า aggregate_tags_count ซึ่งจะรวม new_values (ค่าใหม่) ทั้งหมดสำหรับแต่ละแฮชแท็กและเพิ่มลงในแท็ก total_sum (ผลรวมทั้งหมด) ซึ่งเป็นผลรวมของกลุ่มทั้งหมดและบันทึกข้อมูลใน RDD tags_totals
def aggregate_tags_count(new_values, total_sum): return sum(new_values) + (total_sum or 0) จากนั้นเราจะทำการประมวลผล RDD tags_totals ในแต่ละกลุ่มเพื่อให้สามารถแปลงเป็นตารางชั่วคราวโดยใช้ Spark SQL Context และหลังจากนี้ให้สร้างคำสั่งเพื่อให้สามารถติดแฮชแท็กสิบอันดับแรกกับบัญชีของพวกเขาและใส่ไว้ใน data hashtag_counts_df frame
s corp c corp llc ความแตกต่าง
def get_sql_context_instance(spark_context): if ('sqlContextSingletonInstance' not in globals()): globals()['sqlContextSingletonInstance'] = SQLContext(spark_context) return globals()['sqlContextSingletonInstance'] def process_rdd(time, rdd): print('----------- %s -----------' % str(time)) try: # obtén el contexto spark sql singleton desde el contexto actual sql_context = get_sql_context_instance(rdd.context) # convierte el RDD a Row RDD row_rdd = rdd.map(lambda w: Row(hashtag=w[0], hashtag_count=w[1])) # crea un DF desde el Row RDD hashtags_df = sql_context.createDataFrame(row_rdd) # Registra el marco de data como tabla hashtags_df.registerTempTable('hashtags') # obtén los 10 mejores hashtags de la tabla utilizando SQL e imprímelos hashtag_counts_df = sql_context.sql('select hashtag, hashtag_count from hashtags order by hashtag_count desc limit 10') hashtag_counts_df.show() # llama a este método para preparar los 10 mejores hashtags DF y envíalos send_df_to_dashboard(hashtag_counts_df) except: e = sys.exc_info()[0] print('Error: %s' % e) ขั้นตอนสุดท้ายในแอปพลิเคชัน Spark ของเราคือการส่ง data frame hashtag_counts_df ไปยังแอปแดชบอร์ด ดังนั้นเราจะแปลงเฟรมข้อมูลเป็นสองเมทริกซ์หนึ่งสำหรับแฮชแท็กและอีกอันสำหรับบัญชีของพวกเขา จากนั้นเราจะส่งไปยังแอปแดชบอร์ดผ่าน REST API
def send_df_to_dashboard(df): # extrae los hashtags del marco de data y conviértelos en una matriz top_tags = [str(t.hashtag) for t in df.select('hashtag').collect()] # extrae las cuentas del marco de data y conviértelos en una matriz tags_count = [p.hashtag_count for p in df.select('hashtag_count').collect()] # inicia y envía la data a través de la API REST url = 'http://localhost:5001/updateData' request_data = {'label': str(top_tags), 'data': str(tags_count)} response = requests.post(url, data=request_data) สุดท้ายนี่คือตัวอย่างของผลลัพธ์ของ Spark Streaming ขณะรันและพิมพ์ hashtag_counts_df คุณจะสังเกตเห็นว่าเอาต์พุตถูกพิมพ์ทุกสองวินาทีสำหรับแต่ละช่วงเวลาของกลุ่ม
สร้างแดชบอร์ดแบบเรียลไทม์อย่างง่ายเพื่อแสดงข้อมูล
ตอนนี้เรากำลังจะสร้างแอปพลิเคชันแดชบอร์ดง่ายๆที่ Spark จะได้รับการอัปเดตแบบเรียลไทม์ เรากำลังจะสร้างโดยใช้ Python, Flask และ Charts.js .
ขั้นแรกเราจะสร้างโครงการ Python ด้วยโครงสร้างที่เห็นด้านล่างดาวน์โหลดและเพิ่มไฟล์ Chart.js ในไดเร็กทอรีแบบคงที่
จากนั้นในไฟล์ app.py เราจะสร้างฟังก์ชันที่เรียกว่า update_data ซึ่งจะถูกเรียกโดย Spark ผ่าน URL http://localhost:5001/updateData เพื่อให้สามารถอัปเดตป้ายกำกับส่วนกลางและอาร์เรย์ค่า
หลักการออกแบบเอกภาพตัวอย่าง
ในทำนองเดียวกันฟังก์ชัน refresh_graph_data ถูกสร้างขึ้นเพื่อเรียกโดยคำร้องขอ AJAX เพื่อส่งคืนเลเบลและอาร์เรย์ค่าที่อัปเดตใหม่เป็น JSON ฟังก์ชั่น get_chart_page จะออกจากหน้า chart.html เมื่อเรียก
from flask import Flask,jsonify,request from flask import render_template import ast app = Flask(__name__) labels = [] values = [] @app.route('/') def get_chart_page(): global labels,values labels = [] values = [] return render_template('chart.html', values=values, labels=labels) @app.route('/refreshData') def refresh_graph_data(): global labels, values print('labels now: ' + str(labels)) print('data now: ' + str(values)) return jsonify(sLabel=labels, sData=values) @app.route('/updateData', methods=['POST']) def update_data(): global labels, values if not request.form or 'data' not in request.form: return 'error',400 labels = ast.literal_eval(request.form['label']) values = ast.literal_eval(request.form['data']) print('labels received: ' + str(labels)) print('data received: ' + str(values)) return 'success',201 if __name__ == '__main__': app.run(host='localhost', port=5001) ตอนนี้เราจะสร้างกราฟง่ายๆในไฟล์ chart.html เพื่อให้สามารถแสดงข้อมูลแฮชแท็กและอัปเดตได้แบบเรียลไทม์ ตามที่กำหนดไว้ด้านล่างเราจำเป็นต้องนำเข้าไลบรารี JavaScript, Chart.js และ jquery.min.js.
ในเนื้อความของแท็กเราจำเป็นต้องสร้างผืนผ้าใบและกำหนดรหัสเพื่อให้สามารถอ้างถึงได้ในขณะที่แสดงกราฟเมื่อใช้ JavaScript ในขั้นตอนถัดไป
Top Trending Twitter Hashtags Top Trending Twitter Hashtags
ตอนนี้เราจะสร้างกราฟโดยใช้โค้ด JavaScript ด้านล่าง ขั้นแรกเราใช้องค์ประกอบผ้าใบจากนั้นเราสร้างวัตถุกราฟใหม่และส่งผ่านองค์ประกอบผ้าใบไปที่มันและกำหนดวัตถุข้อมูลดังที่เห็นด้านล่าง
โปรดสังเกตว่าป้ายกำกับข้อมูลจะเชื่อมโยงกับเลเบลและตัวแปรค่าที่ส่งคืนขณะออกจากเพจโดยเรียกแท็ก get_chart_page ในไฟล์ app.py.
ส่วนสุดท้ายคือฟังก์ชันที่กำหนดค่าให้ทำการร้องขอ Ajax ทุก ๆ วินาทีและเรียก URL /refreshData ซึ่งจะดำเนินการ refresh_graph_data ใน app.py และจะส่งคืนข้อมูลที่อัปเดตใหม่จากนั้นอัปเดตกราฟที่ข้อมูลใหม่ออกไป
var ctx = document.getElementById('chart'); var myChart = new Chart(ctx, { type: 'horizontalBar', data: { labels: [{% for item in labels %} '{{item}}', {% endfor %}], datasets: [{ label: '# of Mentions', data: [{% for item in values %} {{item}}, {% endfor %}], backgroundColor: [ 'rgba(255, 99, 132, 0.2)', 'rgba(54, 162, 235, 0.2)', 'rgba(255, 206, 86, 0.2)', 'rgba(75, 192, 192, 0.2)', 'rgba(153, 102, 255, 0.2)', 'rgba(255, 159, 64, 0.2)', 'rgba(255, 99, 132, 0.2)', 'rgba(54, 162, 235, 0.2)', 'rgba(255, 206, 86, 0.2)', 'rgba(75, 192, 192, 0.2)', 'rgba(153, 102, 255, 0.2)' ], borderColor: [ 'rgba(255,99,132,1)', 'rgba(54, 162, 235, 1)', 'rgba(255, 206, 86, 1)', 'rgba(75, 192, 192, 1)', 'rgba(153, 102, 255, 1)', 'rgba(255, 159, 64, 1)', 'rgba(255,99,132,1)', 'rgba(54, 162, 235, 1)', 'rgba(255, 206, 86, 1)', 'rgba(75, 192, 192, 1)', 'rgba(153, 102, 255, 1)' ], borderWidth: 1 }] }, options: { scales: { yAxes: [{ ticks: { beginAtZero:true } }] } } }); var src_Labels = []; var src_Data = []; setInterval(function(){ $.getJSON('/refreshData', { }, function(data) { src_Labels = data.sLabel; src_Data = data.sData; }); myChart.data.labels = src_Labels; myChart.data.datasets[0].data = src_Data; myChart.update(); },1000); เรียกใช้แอปพลิเคชันร่วมกัน
เราจะเรียกใช้แอพพลิเคชั่นสามตัวตามลำดับด้านล่าง: 1. ไคลเอนต์แอพ Twitter 2. Spark App 3. Dashboard Web App
จากนั้นคุณสามารถเข้าถึงแผงควบคุมแบบเรียลไทม์โดยมองหา URL
ตอนนี้คุณสามารถดูกราฟของคุณได้รับการอัปเดตด้านล่าง:
การใช้งาน Apache Streaming ในชีวิตจริง
เราได้เรียนรู้การวิเคราะห์ข้อมูลอย่างง่าย ๆ กับข้อมูลแบบเรียลไทม์โดยใช้ Spark Streaming และรวมเข้ากับแผงควบคุมที่เรียบง่ายโดยใช้บริการเว็บ RESTful จากตัวอย่างนี้เราจะเห็นได้ว่า Spark ทรงพลังเพียงใดเมื่อจับกระแสข้อมูลขนาดใหญ่แปลงร่างและดึงข้อมูลที่มีค่าซึ่งสามารถใช้ในการตัดสินใจได้อย่างง่ายดายในเวลาอันสั้น มีกรณีการใช้งานที่เป็นประโยชน์มากมายซึ่งสามารถนำไปใช้และรองรับอุตสาหกรรมต่างๆเช่นข่าวสารหรือการตลาด
ข้อผิดพลาดของโหนด js ในการจัดการแนวปฏิบัติที่ดีที่สุด
ตัวอย่างอุตสาหกรรมข่าว
เราสามารถติดตามแฮชแท็กที่ถูกกล่าวถึงบ่อยที่สุดเพื่อค้นหาว่าหัวข้อใดที่ผู้คนพูดถึงบนโซเชียลมีเดีย นอกจากนี้เรายังสามารถติดตามแฮชแท็กที่เฉพาะเจาะจงและทวีตของพวกเขาเพื่อค้นหาว่าผู้คนพูดถึงหัวข้อหรือเหตุการณ์ที่เฉพาะเจาะจงในโลกอย่างไร
ตัวอย่างการตลาด
เราสามารถรวบรวมการส่งทวีตและโดยทำการวิเคราะห์ความคิดเห็นจัดหมวดหมู่และกำหนดความสนใจของผู้คนเพื่อนำเสนอข้อเสนอที่เกี่ยวข้องกับความสนใจของพวกเขา
นอกจากนี้ยังมีกรณีการใช้งานมากมายที่สามารถใช้สำหรับการวิเคราะห์โดยเฉพาะ ข้อมูลใหญ่ และสามารถให้บริการได้หลายอุตสาหกรรม สำหรับกรณีการใช้งาน Apache Spark เพิ่มเติมโดยทั่วไปฉันขอแนะนำให้คุณตรวจสอบหนึ่งใน โพสต์ก่อนหน้านี้ .
ฉันขอแนะนำให้คุณอ่านเพิ่มเติมเกี่ยวกับ Spark Streaming ที่นี่ เพื่อเรียนรู้เพิ่มเติมเกี่ยวกับความสามารถและทำการแปลงข้อมูลขั้นสูงสำหรับข้อมูลเพิ่มเติมแบบเรียลไทม์เมื่อใช้งาน
