Streaming tweets with Twitter API - Revision history

Sinoa: /* Stream tweets from sockets into Spark */

2022-09-16T17:41:04Z

Stream tweets from sockets into Spark

← Older revision		Revision as of 17:41, 16 September 2022
Line 82:		Line 82:
	'''IMPORTANT:''' When you debug, the socket will often remain open after your program has crashed, or if you have not closed the Jupyter notebook. So you may have to change PORT number often in the StreamingClient and the test code line (65001, 65002, ...).		'''IMPORTANT:''' When you debug, the socket will often remain open after your program has crashed, or if you have not closed the Jupyter notebook. So you may have to change PORT number often in the StreamingClient and the test code line (65001, 65002, ...).

	== ~~Stream~~ tweets from ~~sockets into~~ Spark ==		== Receive tweets from a socket with streaming Spark ==
	Streaming Spark is the topic of Session 3, but you can run the following script '''spark_receiver.py''' to receive tweets from the socket as a Spark stream (loosely based on [https://sparkbyexamples.com/spark/spark-streaming-from-tcp-socket/ this example]):		Streaming Spark is the topic of Session 3, but you can run the following script '''spark_receiver.py''' to receive tweets from the socket as a Spark stream (loosely based on [https://sparkbyexamples.com/spark/spark-streaming-from-tcp-socket/ this example]):
	import os		import os

Sinoa: /* Stream to Spark */

2022-09-16T17:39:15Z

Stream to Spark

← Older revision		Revision as of 17:39, 16 September 2022
Line 82:		Line 82:
	'''IMPORTANT:''' When you debug, the socket will often remain open after your program has crashed, or if you have not closed the Jupyter notebook. So you may have to change PORT number often in the StreamingClient and the test code line (65001, 65002, ...).		'''IMPORTANT:''' When you debug, the socket will often remain open after your program has crashed, or if you have not closed the Jupyter notebook. So you may have to change PORT number often in the StreamingClient and the test code line (65001, 65002, ...).

	== Stream to Spark ==		== Stream tweets from sockets into Spark ==
	Streaming Spark is the topic of Session 3, but you can run the following script '''spark_receiver.py''' to receive tweets from the socket as a Spark stream (loosely based on [https://sparkbyexamples.com/spark/spark-streaming-from-tcp-socket/ this example]):		Streaming Spark is the topic of Session 3, but you can run the following script '''spark_receiver.py''' to receive tweets from the socket as a Spark stream (loosely based on [https://sparkbyexamples.com/spark/spark-streaming-from-tcp-socket/ this example]):
	import os		import os

Sinoa: /* Stream to Spark */

2022-09-09T13:54:48Z

Stream to Spark

← Older revision		Revision as of 13:54, 9 September 2022
Line 83:		Line 83:

	== Stream to Spark ==		== Stream to Spark ==
	Streaming Spark is the topic of Session 3, but you can run the following script '''spark_receiver.py''' to receive tweets from the socket as a Spark stream:		Streaming Spark is the topic of Session 3, but you can run the following script '''spark_receiver.py''' to receive tweets from the socket as a Spark stream (loosely based on [https://sparkbyexamples.com/spark/spark-streaming-from-tcp-socket/ this example]):
	import os		import os
	import findspark		import findspark
Line 108:		Line 108:
	.awaitTermination()		.awaitTermination()

	~~The code is based on [https://sparkbyexamples~~.~~com/spark/spark-streaming-from-tcp-socket/ this example]:~~		Start '''tweet_harvester.py''' first and then '''spark_receiver.py''' in another window.

	~~Start the streaming tweet harvester~~ first and then ~~start the script with, e.g.:~~
	~~(venv) $ export HOST=localhost PORT=65000; python~~ spark_receiver.py

	Instead of using '''import findspark; findspark.init()''' you can set the '''$SPARK_HOME''' environment variable, for example:		Instead of using '''import findspark; findspark.init()''' you can set the '''$SPARK_HOME''' environment variable, for example:
	export SPARK_HOME=${PWD}/venv/lib/python3.8/site-packages/pyspark		export SPARK_HOME=${PWD}/venv/lib/python3.8/site-packages/pyspark
	Add this line to the end of your '''~/.bashrc''' file if you want to make it permanent. It is safer than '''findspark''' when you have several Spark installations on the same machine.		Add this line to the end of your '''~/.bashrc''' file if you want to make it permanent. It is safer than '''findspark''' when you have several Spark installations on the same machine.

Sinoa: /* Streaming tweets to a socket */

2022-09-09T13:51:23Z

Streaming tweets to a socket

← Older revision		Revision as of 13:51, 9 September 2022
Line 68:		Line 68:
	def on_data(json_data):		def on_data(json_data):
	json_obj = json.loads(json_data.decode())		json_obj = json.loads(json_data.decode())
	~~json_bytes~~ = json.dumps(json_obj['data'])~~.encode()~~		json_str = json.dumps(json_obj['data']['text'])+'\r\n'
	~~connection.sendall(json_bytes~~+'\r\n')		connection.sendall(json_str.encode())
	if tweet_count >= NUM_TWEETS:		if tweet_count >= NUM_TWEETS:
	...close the socket etc...		...close the socket etc...

Sinoa: /* Stream to Spark */

2022-09-09T13:51:03Z

Stream to Spark

← Older revision		Revision as of 13:51, 9 September 2022
Line 88:		Line 88:
	from pyspark.sql import SparkSession		from pyspark.sql import SparkSession

	HOST = ~~os.environ['HOST'] if 'HOST' in os.environ else~~ 'localhost'		HOST = 'localhost'
	PORT = ~~os.environ['PORT'] if 'PORT' in os.environ else '~~65000'		PORT = 65000 # for example
	~~PORT = int(PORT)~~

	findspark.init()		findspark.init()

Sinoa: /* Saving tweets to file */

2022-09-09T13:42:41Z

Saving tweets to file

← Older revision		Revision as of 13:42, 9 September 2022
Line 18:		Line 18:
	(venv) $ pip install tweepy		(venv) $ pip install tweepy

	'''Task 1:''' Run the following minimal boilerplate code to connect to Twitter's API and start downloading a stream of tweets (the code takes a few seconds to connect):		'''Task 1:''' Run the following minimal boilerplate code '''tweet_harvester.py''' to connect to Twitter's API and start downloading a stream of tweets (the code takes a few seconds to connect):
	import json		import json
	import time		import time

Sinoa: /* Streaming tweets to a socket */

2022-09-09T13:41:35Z

Streaming tweets to a socket

← Older revision		Revision as of 13:41, 9 September 2022
Line 49:		Line 49:
	'''Task 5:''' Change the code to write the stream of tweets to a [https://en.wikipedia.org/wiki/Network_socket network socket] instead. Think of a socket as an internal internet connection from your machine back to itself. Different programs on your computer can use this socket to communicate using the regular internet APIs.		'''Task 5:''' Change the code to write the stream of tweets to a [https://en.wikipedia.org/wiki/Network_socket network socket] instead. Think of a socket as an internal internet connection from your machine back to itself. Different programs on your computer can use this socket to communicate using the regular internet APIs.

	Here are some critical pieces of code you need to include in your program:		Here are some critical pieces of code you need to include in your program (loosely based on [https://realpython.com/python-sockets/ this example]):
	import socket		import socket

Line 58:		Line 58:
	tweet_count = 0		tweet_count = 0

	To open a socket for output ~~(as server side~~):		To open a socket for output):
	socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)		socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
	socket.bind((HOST, PORT))		socket.bind((HOST, PORT))
	socket.listen()		socket.listen()
	connection, address = socket.accept()		connection, address = socket.accept()
			Hence, on the output side the program is a server that sends tweets through the socket while, on the input side, it is a client that receives data from Twitter.

	To send data on the socket:		To send data through the socket:
	def on_data(json_data):		def on_data(json_data):
	json_obj = json.loads(json_data.decode())		json_obj = json.loads(json_data.decode())
	json_bytes = json.dumps(json_obj['data']).encode()		json_bytes = json.dumps(json_obj['data']).encode()
	connection.sendall(json_bytes)		connection.sendall(json_bytes+'\r\n')
	if tweet_count >= NUM_TWEETS:		if tweet_count >= NUM_TWEETS:
	...close the socket etc...		...close the socket etc...
	~~The code is based on [https://realpython.com/python-sockets/ this example].~~

			To close the socket on termination:
			socket.close()

	~~You~~ can use the '''nc''' utility to receive data from the socket. From another console window (inside or outside VS Code, a virtual environment is not needed):		For testing, you can use the '''nc''' utility to receive data from the socket. From another console window (inside or outside VS Code, a virtual environment is not needed):
	$ apt install nc		$ apt install nc
	$ nc localhost 65000 # you must start the tweet stream ''before'' you run this command		$ nc localhost 65000 # you must start the tweet stream ''before'' you run this command

	'''IMPORTANT:''' When you debug, the socket will often remain open after your program has crashed, or if you ~~are in~~ the ~~same~~ Jupyter notebook. So you may have to change PORT number often in the StreamingClient and the test code line (65001, 65002, ...).		'''IMPORTANT:''' When you debug, the socket will often remain open after your program has crashed, or if you have not closed the Jupyter notebook. So you may have to change PORT number often in the StreamingClient and the test code line (65001, 65002, ...).

	== Stream to Spark ==		== Stream to Spark ==

Sinoa: /* Saving tweets to file */

2022-09-09T13:32:48Z

Saving tweets to file

← Older revision		Revision as of 13:32, 9 September 2022
Line 20:		Line 20:
	'''Task 1:''' Run the following minimal boilerplate code to connect to Twitter's API and start downloading a stream of tweets (the code takes a few seconds to connect):		'''Task 1:''' Run the following minimal boilerplate code to connect to Twitter's API and start downloading a stream of tweets (the code takes a few seconds to connect):
	import json		import json
	~~from~~ time ~~import sleep~~		import time
	import tweepy		import tweepy

Line 36:		Line 36:
	streaming_client.on_data = on_data		streaming_client.on_data = on_data
	streaming_client.sample(threaded=True)		streaming_client.sample(threaded=True)
	sleep(DURATION)		time.sleep(DURATION)
	streaming_client.disconnect()		streaming_client.disconnect()
	Note how, every time a new tweet is received, tweepy calls the '''on_data()'''-function, which prints the tweeted text to the screen.		Note how, every time a new tweet is received, tweepy calls the '''on_data()'''-function, which prints the tweeted text to the screen. This is just a simple example to get started. In Session 1 we will look at many other ways to access the Twitter API V2 through tweepy.

	'''Task 2:''' ~~Create~~ code ~~that saves~~ the tweets to a file (for example as JSON or CSV). Create a subfolder for tweet files to keep your ''info319-exercise2'' folder tidy.		'''Task 2:''' Change the code to save the tweets to a file (for example as JSON or CSV). Create a subfolder for tweet files to keep your ''info319-exercise2'' folder tidy.

	'''Task 3:''' Change the code so you download and save additional information about each tweet, for example the handle (user name) of the tweeter, whether it is a retweet of another id, and perhaps the date and time.		'''Task 3:''' Change the code so you download and save additional information about each tweet, for example the handle (user name) of the tweeter, whether it is a retweet of another id, and perhaps the date and time.

	'''Task 4:''' Refactor the code to prepare for cleaner termination~~. Remove the these lines from the~~ '''~~run~~()'''-function ~~(along with '''threaded=True''' in~~ the ~~call to~~ '''~~sample~~()''' and ~~along with everything about the '''DURATION''' variable):~~		'''Task 4:''' Refactor the code to prepare for cleaner termination by creating a new '''on_finish()'''-function, which disconnects the streaming client, closes open files, and performs any other needed termination tasks. Remove '''time.sleep(DURATION)''', and instead make the '''on_data()'''-function call '''on_finish()''' when a given number of tweets (say 100 to start) has been received.
	~~sleep(DURATION)~~
	~~streaming_client.disconnect()~~
	~~Instead,~~ make the '''on_data()'''-function call ~~a new~~ '''on_finish()'''~~-function~~ when a given number of tweets (say 100 to start) has been received~~. Your '''on_finish()'''-function should disconnect the stream, close the file, and do any other termination tasks~~.

	== Streaming tweets to a socket ==		== Streaming tweets to a socket ==

Sinoa: /* Saving tweets to file */

2022-09-09T13:26:47Z

Saving tweets to file

← Older revision		Revision as of 13:26, 9 September 2022
Line 15:		Line 15:

	== Saving tweets to file ==		== Saving tweets to file ==
			Install the tweepy API:
	(venv) $ pip install tweepy		(venv) $ pip install tweepy

Sinoa: /* Get Twitter API account */

2022-09-09T13:25:57Z

Get Twitter API account

← Older revision		Revision as of 13:25, 9 September 2022
Line 12:		Line 12:
	(venv) $ echo "...the long alpha-numeric token string goes here..." > keys/bearer_token		(venv) $ echo "...the long alpha-numeric token string goes here..." > keys/bearer_token

	If you use git, you can add the line '''/keys/''' to your ''.gitignore' file as well. You do not want it uploaded to an external repository.		If you use git, you can add the line '''/keys/''' to your ''.gitignore'' file as well. You do not want it uploaded to an external repository.

	== Saving tweets to file ==		== Saving tweets to file ==