Wsrtjtyk

I am learning Structured Streaming with Databricks and I'm struggling with the DataStreamWriter console mode.

My program:

• Simulates the streaming arrival of files to the folder "monitoring_dir" (one new file is transferred from "source_dir" each 10 seconds).


• Uses a DataStreamReader to populate the Unbounded DataFrame "inputUDF" with the content of each new file.


• Uses a DataStreamWriter to output the new rows of "inputUDF" to a valid sink.

Whereas the program works when choosing to use a File sink (the batches are appended to text-format files in "result_dir"), I cannot see anything displayed when choosing Console sink.

Moreover, when I run the equivalent version of the program in my local machine (with Spark installed on it) it works fine both for File and Console sinks.

My question is:

• 
  How can I make this program to output to Console sink and display the results when using Databricks?

Thank you very much in advance!

Best regards,
Nacho

My Program: myTest.py

import pyspark

import pyspark.sql.functions



import time



#------------------------------------

# FUNCTION get_source_dir_file_names

#------------------------------------ 

def get_source_dir_file_names(source_dir):



    # 1. We create the output variable

    res = 



    # 2. We get the FileInfo representation of the files of source_dir

    fileInfo_objects = dbutils.fs.ls(source_dir)



    # 3. We traverse the fileInfo objects, to get the name of each file

    for item in fileInfo_objects:      

        # 3.1. We get a string representation of the fileInfo

        file_name = str(item)



        # 3.2. We look for the pattern name= to remove all useless info from the start

        lb_index = file_name.index("name='")

        file_name = file_name[(lb_index + 6):]



        # 3.3. We look for the pattern ') to remove all useless info from the end

        ub_index = file_name.index("',")

        file_name = file_name[:ub_index]



        # 3.4. We append the name to the list

        res.append(file_name)



    # 4. We sort the list in alphabetic order

    res.sort()



    # 5. We return res

    return res



#------------------------------------

# FUNCTION streaming_simulation

#------------------------------------ 

def streaming_simulation(source_dir, monitoring_dir, time_step_interval):

    # 1. We get the names of the files on source_dir

    files = get_source_dir_file_names(source_dir)



    # 2. We get the starting time of the process

    time.sleep(time_step_interval * 0.1)



    start = time.time()



    # 3. We set a counter in the amount of files being transferred

    count = 0



    # 4. We simulate the dynamic arriving of such these files from source_dir to dataset_dir

    # (i.e, the files are moved one by one for each time period, simulating their generation).

    for file in files:

        # 4.1. We copy the file from source_dir to dataset_dir#

        dbutils.fs.cp(source_dir + file, monitoring_dir + file)



        # 4.2. We increase the counter, as we have transferred a new file

        count = count + 1



        # 4.3. We wait the desired transfer_interval until next time slot.

        time.sleep((start + (count * time_step_interval)) - time.time())



    # 5. We wait a last time_step_interval

    time.sleep(time_step_interval)



#------------------------------------

# FUNCTION my_main

#------------------------------------ 

def my_main():

    # 0. We set the mode

    console_sink = True



    # 1. We set the paths to the folders

    source_dir = "/FileStore/tables/my_dataset/"

    monitoring_dir = "/FileStore/tables/my_monitoring/"

    checkpoint_dir = "/FileStore/tables/my_checkpoint/"

    result_dir = "/FileStore/tables/my_result/"



    dbutils.fs.rm(monitoring_dir, True)

    dbutils.fs.rm(result_dir, True)

    dbutils.fs.rm(checkpoint_dir, True)  



    dbutils.fs.mkdirs(monitoring_dir)

    dbutils.fs.mkdirs(result_dir)

    dbutils.fs.mkdirs(checkpoint_dir)    



    # 2. We configure the Spark Session

    spark = pyspark.sql.SparkSession.builder.getOrCreate()

    spark.sparkContext.setLogLevel('WARN')    



    # 3. Operation C1: We create an Unbounded DataFrame reading the new content copied to monitoring_dir

    inputUDF = spark.readStream.format("text")

                               .load(monitoring_dir)



    myDSW = None

    # 4. Operation A1: We create the DataStreamWritter...



    # 4.1. To either save to result_dir in append mode  

    if console_sink == False:

        myDSW = inputUDF.writeStream.format("text")

                                    .option("path", result_dir) 

                                    .option("checkpointLocation", checkpoint_dir)

                                    .trigger(processingTime="10 seconds")

                                    .outputMode("append")   

    # 4.2. Or to display by console in append mode    

    else:

        myDSW = inputUDF.writeStream.format("console")

                                    .trigger(processingTime="10 seconds")

                                    .outputMode("append")   



    # 5. We get the StreamingQuery object derived from starting the DataStreamWriter

    mySQ = myDSW.start()



    # 6. We simulate the streaming arrival of files (i.e., one by one) from source_dir to monitoring_dir

    streaming_simulation(source_dir, monitoring_dir, 10)



    # 7. We stop the StreamingQuery to finish the application

    mySQ.stop()    



#-------------------------------

# MAIN ENTRY POINT

#-------------------------------strong text

if __name__ == '__main__':

    my_main()

My Dataset: f1.txt

First sentence.

Second sentence.

My Dataset: f2.txt

Third sentence.

Fourth sentence.

My Dataset: f3.txt

Fifth sentence.

Sixth sentence.