<jupyter_start><jupyter_text>Create the SparkSession and set the environment to use our local MongoDB cluster<jupyter_code>from pyspark.sql import SparkSession

spark = SparkSession.\
        builder.\
        appName("pyspark-notebook2").\
        master("spark://spark-master:7077").\
        config("spark.executor.memory", "512m").\
        config("spark.mongodb.input.uri","mongodb://mongo1:27017/Stocks.Source").\
        config("spark.mongodb.output.uri","mongodb://mongo1:27017/Stocks.Source").\
        config("spark.jars.packages", "org.mongodb.spark:mongo-spark-connector_2.12:3.0.0").\
        getOrCreate()<jupyter_output><empty_output><jupyter_text>Next load the dataframes from MongoDB<jupyter_code>df = spark.read.format("mongo").load()<jupyter_output><empty_output><jupyter_text>Let’s verify the data was loaded by looking at the schema:<jupyter_code>df.printSchema()<jupyter_output><empty_output><jupyter_text>We can see that the tx_time field is loaded as a string. We can easily convert this to a time by issuing a cast statement:<jupyter_code>df = df.withColumn("tx_time", df.tx_time.cast("timestamp"))<jupyter_output><empty_output><jupyter_text>Next, we can add a new ‘movingAverage’ column that will show a moving average based upon the previous value in the dataset. To do this we leverage the PySpark Window function as follows:<jupyter_code>from pyspark.sql import functions as F
from pyspark.sql.window import Window

movAvg = df.withColumn("movingAverage", F.avg("price")
             .over( Window.partitionBy("company_symbol").rowsBetween(-1,1)) )<jupyter_output><empty_output><jupyter_text>To see our data with the new moving average column we can issue a movAvg.show().<jupyter_code>movAvg.show()<jupyter_output><empty_output><jupyter_text>To update the data in our MongoDB cluster, we use the save method.<jupyter_code>movAvg.write.format("mongo").option("replaceDocument", "true").mode("append").save()<jupyter_output><empty_output><jupyter_text>We can also use the power of the MongoDB Aggregation Framework to pre-filter, sort or aggregate our MongoDB data.<jupyter_code>pipeline = "[{'$group': {_id:'$company_name', 'maxprice': {$max:'$price'}}},{$sort:{'maxprice':-1}}]"
aggPipelineDF = spark.read.format("mongo").option("pipeline", pipeline).option("partitioner", "MongoSinglePartitioner").load()
aggPipelineDF.show()<jupyter_output><empty_output><jupyter_text>Finally we can use SparkSQL to issue ANSI-compliant SQL against MongoDB data as follows:<jupyter_code>movAvg.createOrReplaceTempView("avgs")
sqlDF=spark.sql("SELECT * FROM avgs WHERE movingAverage > 43.0")
sqlDF.show()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.131797583081571, 'fraction_numerical': 0.012084592145015106, 'mean_word_length': 6.257534246575342, 'pattern_counts': {'":': 0, '<': 37, '<?xml version=': 0, '>': 38, 'https://': 0, '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': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '415326', 'license': 'MIT', 'n_tokens_mistral': 908, 'n_tokens_neox': 858, 'n_words': 191, 'path': 'work/examples/.ipynb_checkpoints/pyspark-mongo-checkpoint.ipynb', 'repo_name': 'stellarshenson/stellars-dataplatform', 'size': '2648'}