write_to_delta_lake

from pyspark.sql import SparkSession
from pyspark.sql.types import *
from pyspark.sql.functions import *
from delta import *



spark = SparkSession \
    .builder \
    .appName("word_counter") \
    .config('spark.jars.packages','io.delta:delta-core_2.12:2.1.0')  \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate() 


inputPath = "./text.txt"
schema1 = StructType([StructField('word', StringType(), True)])    

print('step1')
inputDF = spark.read.option("delimiter", "/t").schema(schema1).csv(inputPath)
inputDF.show()

print('step2')
inputDF = inputDF.select(split(col("word")," ").alias("word"))
inputDF.show()

print('step3')
inputDF = inputDF.select(inputDF.word,explode(inputDF.word)).withColumnRenamed("col","unique_word")
inputDF.show()  

print('step4')
inputDF = inputDF.filter(inputDF["unique_word"] != "")
inputDF.show()  

print('step5')
inputDF = inputDF.groupBy("unique_word").count()    
inputDF.show()


print('writing to delta')
inputDF.write.mode("overwrite").format("delta").save("/tmp/delta-table")
DF_from_Delta = spark.read.format("delta").load("/tmp/delta-table")
DF_from_Delta.show()