<jupyter_start><jupyter_text>Spark_practice<jupyter_code>import os
from pyspark import SparkConf, SparkContext, SQLContext
from pyspark.sql import SparkSession
from pyspark.sql.functions import udf, length, when, col
from pyspark.sql.types import BooleanType, IntegerType, LongType, StringType, ArrayType, FloatType, StructType, StructField
import pyspark.sql.functions as F
from pyspark.sql.functions import pandas_udf
from pyspark.sql.functions import PandasUDFType
from jinja2 import Environment, FileSystemLoader
import pandas as pd

os.environ["PYSPARK_PYTHON"] = "/opt/conda/bin/python"
# setting constants
SPARK_ADDRESS = "###############"

APP_NAME = "Spark_practice"
NORMALIZED_APP_NAME = APP_NAME.replace('/', '_').replace(':', '_')

APPS_TMP_DIR = os.path.join(os.getcwd(), "tmp")
APPS_CONF_DIR = os.path.join(os.getcwd(), "conf")
APPS_LOGS_DIR = os.path.join(os.getcwd(), "logs")
LOG4J_PROP_FILE = os.path.join(APPS_CONF_DIR, "pyspark-log4j-{}.properties".format(NORMALIZED_APP_NAME))
LOG_FILE = os.path.join(APPS_LOGS_DIR, 'pyspark-{}.log'.format(NORMALIZED_APP_NAME))
EXTRA_JAVA_OPTIONS = "-Dlog4j.configuration=file://{} -Dspark.hadoop.dfs.replication=1"\
    .format(LOG4J_PROP_FILE)

LOCAL_IP = socket.gethostbyname(socket.gethostname())
spark.stop()
# preparing configuration files from templates
for directory in [APPS_CONF_DIR, APPS_LOGS_DIR, APPS_TMP_DIR]:
    if not os.path.exists(directory):
        os.makedirs(directory)

env = Environment(loader=FileSystemLoader('/opt'))
template = env.get_template("pyspark_log4j.properties.template")
template\
    .stream(logfile=LOG_FILE)\
    .dump(LOG4J_PROP_FILE)
# run spark
spark = SparkSession\
    .builder\
    .appName(APP_NAME)\
    .master(SPARK_ADDRESS)\
    .config('spark.ui.port', "####")\
    .config("spark.driver.host", LOCAL_IP)\
    .config("spark.memory.fraction", "0.8")\
    .config("spark.memory.storageFraction", "0.6")\
    .config("spark.executor.instances", "5")\
    .config("spark.executor.cores", '4')\
    .config("spark.executor.memory", "5g")\
    .config("spark.driver.memory", "4g")\
    .config("spark.driver.extraJavaOptions", EXTRA_JAVA_OPTIONS)\
    .config("spark.executor.memory", "6g")\
    .config("spark.kubernetes.namespace", "#########")\
    .config("spark.kubernetes.driver.label.appname", APP_NAME)\
    .config("spark.kubernetes.executor.label.appname", APP_NAME)\
    .config("spark.kubernetes.container.image.pullPolicy", "Always")\
    .config("spark.kubernetes.container.image", "node##.st:###/spark-executor:########")\
    .config("spark.kubernetes.executor.deleteOnTermination", "true")\
    .config("spark.local.dir", "/tmp/spark")\
    .getOrCreate()
posts_df = spark.read.json("hdfs:///shared/bigdata20/posts_api.json")
posts_likes_df = spark.read.parquet("hdfs:///shared/bigdata20/posts_likes.parquet")
followers_df = spark.read.parquet("hdfs:///shared/bigdata20/followers.parquet")
followers_posts_df = spark.read.json("hdfs:///shared/bigdata20/followers_posts_api_final.json")
followers_posts_likes_df = spark.read.parquet("hdfs:///shared/bigdata20/followers_posts_likes.parquet")
posts_df.printSchema()<jupyter_output>root
 |-- attachments: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- album: struct (nullable = true)
 |    |    |    |-- created: long (nullable = true)
 |    |    |    |-- description: string (nullable = true)
 |    |    |    |-- id: string (nullable = true)
 |    |    |    |-- owner_id: long (nullable = true)
 |    |    |    |-- size: long (nullable = true)
 |    |    |    |-- thumb: struct (nullable = true)
 |    |    |    |    |-- access_key: string (nullable = true)
 |    |    |    |    |-- album_id: long (nullable = true)
 |    |    |    |    |-- date: long (nullable = true)
 |    |    |    |    |-- id: long (nullable = true)
 |    |    |    |    |-- owner_id: long (nullable = true)
 |    |    |    |    |-- sizes: array (nullable = true)
 |    |    |    |    |    |-- element: struct (containsNull = true)
 |    |    |    |    |    |    |-- height: long (nullable = true)
 |    |    |    |    |    |    |-- type: string (nullable = true)
 |   [...]<jupyter_text>Top 20 posts in the group by likes<jupyter_code>likes = posts_df.select('id', 'likes.count')\
            .orderBy("likes", ascending = False)\
            .limit(20)

likes.show()<jupyter_output>+-----+-----+
|   id|count|
+-----+-----+
|32022| 1637|
|35068| 1629|
|17492| 1516|
|18526| 1026|
|19552|  955|
|41468|  952|
|19419|  868|
|29046|  824|
|32546|  786|
|24085|  765|
|40180|  759|
|33658|  708|
|13532|  633|
|40842|  631|
|35117|  588|
|17014|  581|
|19583|  553|
|19809|  552|
|27455|  550|
|11999|  549|
+-----+-----+<jupyter_text>Top 20 posts in the group by comments<jupyter_code>comments = posts_df.select('id', 'comments.count')\
                .orderBy("comments", ascending = False)\
                .limit(20)

comments.show()<jupyter_output>+-----+-----+
|   id|count|
+-----+-----+
|24085|  850|
|22540|  250|
|27722|  192|
| 8285|  148|
|26860|  113|
|13571|  107|
|39294|  104|
|36680|   96|
|41739|   92|
|26006|   92|
|12426|   91|
|21499|   88|
|39163|   83|
|39407|   83|
|11267|   81|
|31548|   80|
|11158|   70|
|39082|   67|
|12687|   61|
|14602|   61|
+-----+-----+<jupyter_text>Top 20 posts in the group by reposts<jupyter_code>reposts = posts_df.select('id', 'reposts.count')\
                .orderBy("reposts", ascending = False)\
                .limit(20)

reposts.show()<jupyter_output>+-----+-----+
|   id|count|
+-----+-----+
|17492|  334|
|19552|  246|
|32022|  210|
|11842|  129|
|19419|  126|
|13532|  110|
|17014|  105|
|35068|  101|
|41266|   92|
|12593|   90|
|29046|   87|
|41468|   85|
|11999|   85|
|19809|   84|
|17167|   81|
|10833|   78|
|18543|   77|
|16596|   76|
|18156|   74|
|37262|   71|
+-----+-----+<jupyter_text>Top 20 users by likes<jupyter_code>users_likes= posts_likes_df.groupby('likerId')\
                    .agg(F.count('itemId').name('likes'))\
                    .orderBy('likes', ascending = False)\
                    .limit(20)

users_likes.show()<jupyter_output>+---------+-----+
|  likerId|likes|
+---------+-----+
|  2070090| 4801|
|  2397858| 2055|
|  1475301| 1829|
|    18239| 1569|
|   546612| 1245|
|     6371|  907|
|  1841959|  746|
| 78440957|  709|
|   120248|  699|
| 40981497|  611|
|    22158|  553|
|207628162|  548|
|329377723|  504|
| 76071304|  474|
| 14805173|  440|
|   317799|  385|
| 56355640|  375|
| 52042971|  338|
|  7437271|  336|
|136506644|  335|
+---------+-----+<jupyter_text>Top 20 users by reposts<jupyter_code>users_reposts = posts_df.where(col('copy_history.owner_id').getItem(0) > 0)\
                            .groupby(col('copy_history.owner_id').name('user_id'))\
                            .agg(F.count(col('copy_history')).name('reposts_count'))\
                            .orderBy('reposts_count', ascending = False)\
                            .limit(20)

users_reposts.show()<jupyter_output>+-----------+-------------+
|    user_id|reposts_count|
+-----------+-------------+
|    [18239]|           20|
|     [1398]|            3|
|   [252574]|            2|
| [53083705]|            2|
| [54314207]|            1|
|   [351122]|            1|
| [97496712]|            1|
|[140109164]|            1|
|   [190868]|            1|
| [37885305]|            1|
|  [1155551]|            1|
| [92580857]|            1|
|  [1475301]|            1|
| [40931681]|            1|
| [26167986]|            1|
|  [9718166]|            1|
|  [2070090]|            1|
| [14170688]|            1|
|  [3716921]|            1|
|[185886061]|            1|
+-----------+-------------+<jupyter_text>Reposts of the original posts of the itmo group from user posts<jupyter_code>itmo_posts = followers_posts_df\
                .select('owner_id', col('copy_history')['id'].getItem(0).name('original_post_id'))\
                .where(col('copy_history')['owner_id'].getItem(0) == -94)\
                .groupby('original_post_id')\
                .agg(F.collect_list('owner_id').name('user_ids'))\
                .limit(10)

itmo_posts.show()<jupyter_output>+----------------+--------------------+
|original_post_id|            user_ids|
+----------------+--------------------+
|           39259|[15900015, 159562...|
|           40084|         [268247082]|
|           41535|         [342841693]|
|           39515|[4197185, 1584029...|
|           40100| [317799, 483010700]|
|           42234|         [180907432]|
|           40180|[268247082, 14834...|
|           41659|  [86002, 153941761]|
|           40839|           [1548876]|
|       456239414|[471165, 4068532,...|
+----------------+--------------------+<jupyter_text>Emoticons in posts<jupyter_code>from pyspark.sql.types import ArrayType, StringType
!{sys.executable} -m pip install --user --trusted-host pypi-registry.supplementary-services.svc.cluster.local --index http://pypi-registry.supplementary-services.svc.cluster.local:#### ####    
import emoji
import re
import string
@udf(returnType=ArrayType(StringType())) 

def emojies(text):
    if (not text):
        return
    import string
    text = text.translate(str.maketrans('', '', string.punctuation))
    text = emoji.demojize(text)
    text = re.findall(r'(:[^:]*:)', text)
    list_emoji = [emoji.emojize(x) for x in text]
    return list_emoji

emojies_in_posts = posts_df.select("id", 'text')\
                    .withColumn("emojies", emojies(col("text")))\
                    .where("size(emojies) > 0")\
                    .select("id", F.explode("emojies").name("emoji"))\
                    .groupBy("emoji").agg(F.count("id").name("count"),F.countDistinct("id").name("frequency"),)\
                    .withColumn("avg_cnt_per_post", col("count") / col("frequency"))\
                    .withColumn("difference", col("count") - col("frequency"))\
                    .limit(5)


emojies_in_posts.show()<jupyter_output>+-----+-----+---------+----------------+----------+
|emoji|count|frequency|avg_cnt_per_post|difference|
+-----+-----+---------+----------------+----------+
|   📸|    6|        6|             1.0|         0|
|   📷|    3|        1|             3.0|         2|
|   🛸|    1|        1|             1.0|         0|
|   🎮|    3|        3|             1.0|         0|
|   🌎|    4|        4|             1.0|         0|
+-----+-----+---------+----------------+----------+<jupyter_text>Top 10 most popular emoticons<jupyter_code>most_popular_emojies = emojies_in_posts.select("emoji", "count")\
                        .orderBy(F.desc("count"))\
                        .limit(5)

most_popular_emojies.show()<jupyter_output>+-----+-----+
|emoji|count|
+-----+-----+
|   🕞|    3|
|   🍂|    3|
| 👩🏻|    1|
|   🍜|    1|
|   📊|    1|
+-----+-----+<jupyter_text>Top 5 emoticons which have the greatest difference between their overall count and frequency<jupyter_code>top_5_emojies = emojies_in_posts.select("emoji", "difference")\
                    .orderBy(F.desc("difference"))\
                    .limit(5)

top_5_emojies.show()<jupyter_output>+-----+----------+
|emoji|difference|
+-----+----------+
|   🔎|         4|
|   🌟|         0|
| 🎅🏼|         0|
|    ⛳|         0|
|   📢|         0|
+-----+----------+<jupyter_text>Top 5 emoticons with average count per post<jupyter_code>top_5 = emojies_in_posts.select("emoji", "avg_cnt_per_post")\
            .orderBy(F.desc("avg_cnt_per_post"))\
            .limit(5)

top_5.show()<jupyter_output>+-----+------------------+
|emoji|  avg_cnt_per_post|
+-----+------------------+
|   🔎|2.3333333333333335|
|   🌟|               1.0|
|   📆|               1.0|
|    ⌚|               1.0|
|   📢|               1.0|
+-----+------------------+<jupyter_text>Probable “fans”<jupyter_code>from pyspark.sql.window import Window

probable_fan = followers_posts_likes_df.select('ownerId', 'likerId', 'itemId')\
                    .groupby('likerId', 'ownerId')\
                    .agg(F.count('itemId').name('likes_num'))\
                    .orderBy('likes_num')\
                    .withColumn('number', F.row_number().over(Window.partitionBy('likerId').orderBy(F.desc('likes_num'))))

probable_fans = probable_fan.select('likerId','ownerId').where(probable_fan.number <= 10).orderBy('likerId', 'number')

probable_fans.show(10)<jupyter_output>+-------+-------+
|likerId|ownerId|
+-------+-------+
|      9| 654356|
|     14|1986125|
|     14|  35524|
|     14|   3420|
|     14|3680017|
|     15|1227412|
|     15| 473831|
|     17|1986125|
|     17|   3420|
|     17|  88060|
+-------+-------+
only showing top 10 rows<jupyter_text>Probable friends<jupyter_code>probable_friends = probable_fans.alias("z1")\
.join(probable_fans.alias("z2"), (col('z2.likerId') == col('z1.ownerId')), 'inner')\
.select(col('z1.likerId'), col('z1.ownerId'), col('z2.ownerId').alias('ownerId2'))\
.where((col('likerId') == col('ownerId2')) & (col('likerId') != col('ownerId')))\
.select(col('likerId'),col('ownerId')).distinct().limit(10)

probable_friends.show()<jupyter_output>+---------+---------+
|  likerId|  ownerId|
+---------+---------+
|   370123|   588552|
| 49848562|176592944|
| 63394933|163435990|
| 63394933|267810099|
| 80734900|306995029|
| 90898752|168438070|
|153123605|334837381|
|153123605|405143020|
|153123605|349304983|
|182703092|158548069|
+---------+---------+
{'fraction_non_alphanumeric': 0.2205474315710536, 'fraction_numerical': 0.10446194225721785, 'mean_word_length': 2.268627450980392, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 42, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 1, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '13', 'dir': 'jupyter-structured-clean-dedup', 'id': '144553', 'license': 'MIT', 'n_tokens_mistral': 5920, 'n_tokens_neox': 5037, 'n_words': 785, 'path': 'Spark_practice/Spark_practice.ipynb', 'repo_name': 'SimonGreenstain/My_works', 'size': '13335'}