<jupyter_start><jupyter_text>Direct Messaging<jupyter_code>api.send_direct_message()<jupyter_output><empty_output><jupyter_text>Follower operations<jupyter_code># to follow a given profile
api.create_friendship("TensorFlow")
# to follow or unfollow a given profile
api.destroy_friendship("TensorFlow")
api.followers_ids("TensorFlow")[:3]
api.trends_available()<jupyter_output><empty_output><jupyter_text>Real Time Twitter Streaming<jupyter_code>import json
from pathlib import Path

public_tweets = api.home_timeline()
for tweet in public_tweets:
    base= Path("twitter-logs")
    jsonpath = base / (tweet.id_str + ".json")
    base.mkdir(exist_ok=True)
    jsonpath.write_text(json.dumps(tweet._json))
IN_PATH = "./twitter-logs/"
OUT_PATH = ""
timestampformat = "EEE MMM dd HH:mm:ss zzzz yyyy"

spark = SparkSession.builder.appName("Twitter-Streaming").getOrCreate()
static= spark.read.json(IN_PATH)
schema = static.limit(100).schema
streaming = spark.readStream.schema(schema).option("maxFilesPerTrigger", 10)\
.json("./twitter-logs/")
static.printSchema()
# examining various counts of tweets grouped by user
tweet_counts = streaming.groupBy("user").count()

tweet_query = tweet_counts.writeStream.queryName("tweet_counts")\
.format("memory").outputMode("complete")\
.start()
spark.sql("SELECT * FROM tweet_counts").show()
selection_query = streaming\
.select("user", "user.favourites_count", "user.followers_count", 
        "user.friends_count", "user.statuses_count")\
.writeStream\
.queryName("tweet_metrics")\
.format("memory")\
.outputMode("append")\
.start()
spark.sql("SELECT * FROM tweet_metrics").show()
selection_query = streaming\
.select("user.id", "user.favourites_count", "user.followers_count", 
        "user.friends_count", "user.statuses_count")\
.writeStream\
.queryName("tweet_metrics_agg")\
.format("memory")\
.outputMode("append")\
.start()
spark.sql("SELECT * FROM tweet_metrics_agg").show(truncate= False)
selection_query = streaming\
.select("user.id", "user.favourites_count", "user.followers_count", 
        "user.friends_count", "user.statuses_count")\
.groupBy("id").count()
selection_query\
.writeStream\
.queryName("tweet_metrics_avg")\
.format("memory")\
.outputMode("complete")\
.start()
spark.sql("SELECT * FROM tweet_metrics_avg").show(truncate= False)
# Examine the streaming queries which are active
spark.streams.active
selection_query = streaming\
.select("user", "retweet_count", "lang", "created_at")\
.writeStream\
.queryName("selection_query_")\
.format("memory")\
.outputMode("append")\
.start()
spark.sql("SELECT * FROM selection_query_").show()<jupyter_output>+--------------------+-------------+----+--------------------+
|                user|retweet_count|lang|          created_at|
+--------------------+-------------+----+--------------------+
|{false, Sat Oct 3...|            0|  en|Sun Jun 06 23:50:...|
|{false, Tue Feb 1...|            5|  en|Sun Jun 06 23:39:...|
|{false, Sun Apr 0...|            8|  en|Sun Jun 06 23:45:...|
|{false, Wed Apr 2...|            0|  en|Sun Jun 06 23:40:...|
|{false, Mon Mar 0...|            1|  en|Sun Jun 06 23:39:...|
|{false, Wed Apr 2...|            2|  en|Sun Jun 06 23:50:...|
|{false, Fri Aug 2...|            0|  en|Sun Jun 06 23:45:...|
|{false, Fri Aug 2...|            0|  en|Sun Jun 06 23:50:...|
|{false, Fri Aug 2...|            0|  en|Sun Jun 06 23:40:...|
|{false, Fri Feb 1...|            0|  en|Sun Jun 06 23:50:...|
|{false, Tue Aug 2...|            8|  en|Sun Jun 06 23:27:...|
|{false, Fri May 0...|            3|  en|Sun Jun 06 23:30:...|
|{false, Sun Apr 0...|           12|  en|Sun Jun 06 23:[...]<jupyter_text>Caching<jupyter_code>start= time.time()
sales = spark.read.csv('online-retail-dataset.csv', header= True, inferSchema= True)
sales_country= sales.groupBy("Country").count().collect()
sales_customer= sales.groupBy("CustomerID").count().collect()
sales_invoice= sales.groupBy("InvoiceNo").count().collect()
end= time.time()
print("total time elapsed: ", end-start, "seconds")
sales.cache()
sales.count()
start= time.time()
sales_country= sales.groupBy("Country").count().collect()
sales_customer= sales.groupBy("CustomerID").count().collect()
sales_invoice= sales.groupBy("InvoiceNo").count().collect()
end= time.time()
print("total time elapsed: ", end-start, "seconds")
?sales.persist
from pyspark.storagelevel import StorageLevel
sales.persist(storageLevel= StorageLevel.DISK_ONLY)
sales.count()<jupyter_output><empty_output><jupyter_text>Twitter Default Streaming<jupyter_code>import json
import tweepy

class MyStreamListener(tweepy.StreamListener):
    def __init__(self, api):
        self.api = api
        self.me = api.me()

    def on_status(self, tweet):
        print(f"{tweet.user.name}:{tweet.text}")

    def on_error(self, status):
        print("Error detected")

# Authenticate to Twitter
auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
auth.set_access_token(ACCESS_TOKEN, ACCESS_SECRET)

# Create API object
api = tweepy.API(auth, wait_on_rate_limit=True,
    wait_on_rate_limit_notify=True)

tweets_listener = MyStreamListener(api)
stream = tweepy.Stream(api.auth, tweets_listener)
stream.filter(track=["Covid-19"], languages=["en"])<jupyter_output>vine:RT @WanmingYang: Respeitar a ciência é respeitar e cuidar da vida.
Ukrinform:У Штатах почали вакцинувати тварин проти COVID-19
https://t.co/3hqjbb8GaV https://t.co/X8XQPWZyBS
Arthur.:RT @SteveBakerHW: 🤔The NHS COVID-19 official contact tracing app:
Fighting GQP Fascists & Traitors one day at a time:CNN medical analyst  Dr. <NAME>  fires back at Tucker Carlson with damning questions about his segments on… https://t.co/Hpm08QOb8A
OFFICIALGIGICAPONE:RT @GiGiCaponePR: NEWS: 'Tell all your TikTok buddies to get vaccinated': Fauci finds a new way to fight Covid-19 https://t.co/lCWEytICST V…
<NAME>:RT @GregBrooke4: @BevanLab @NatureComms @DrDALEACH @stathisgiotis @wendybarclay11 @RZwacka @DrAndreaMohr1 Following our recent Nature Comms…
Kuro_Unplugged:Changes in sleep phase and body weight of mobile health App users during COVID-19 mild lockdown in Japan… https://t.co/lEAgktI6lO
Praveen JR:RT @gerdosi: SARS-CoV-2 binds to human epithelial cells and causes permeability.

SAR[...]
{'fraction_non_alphanumeric': 0.17, 'fraction_numerical': 0.021129032258064517, 'mean_word_length': 4.176126878130217, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 21, 'https://': 5, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '472170', 'license': 'MIT', 'n_tokens_mistral': 2304, 'n_tokens_neox': 2128, 'n_words': 480, 'path': 'RealTimeDataAnalysis/Lecture10/Lecture-10.ipynb', 'repo_name': 'robretoarenal/BTS2', 'size': '6214'}