spark-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Mich Talebzadeh <mich.talebza...@gmail.com>
Subject Re: Delivery Status Notification (Failure)
Date Sun, 02 May 2021 09:45:42 GMT
This message is in two parts

Hi,

I did some tests on these. The idea being running  Spark Structured
Streaming (SSS) on a collection of records since the last run of SSS and
shutdown SSS job.

Some parts of this approach has been described in the following databricks
blog

Running Streaming Jobs Once a Day For 10x Cost Savings
<https://databricks.com/blog/2017/05/22/running-streaming-jobs-day-10x-cost-savings.html>
However, real life is more complicated than that.

Let us look at a typical example as depicted in my lousy diagram attached

sources (trail files) --> Kafka --> Flume --> write to Cloud storage
(mounted locally) --> SSS --> BigQuery


What I have in here is a typical example of *trail files* produced by
source. This could be some CDC like Oracle Golden Gate sending
committed logs or anything that writes to files. We use Kafka to ingest
these files and we use Apache flume
<http://flume.apache.org/#:~:text=Welcome%20to%20Apache%20Flume%20%C2%B6%20Flume%20is%20a,reliability%20mechanisms%20and%20many%20failover%20and%20recovery%20mechanisms.>
to
move these files onto Google Cloud Storage (gs:// ) *mounted as a local
file system* in the edge node through Cloud Storage Fuse
<https://cloud.google.com/storage/docs/gcs-fuse>


The advantage of these storage types is that both on-premise and cloud
applications can take care of it.


For on-premise the mount point would be say data_path =
"file:///mnt/gs/prices/data/" where gs://etcbucket is mounted as /mnt/gs


For cloud reference it would be data_path = "etcbucket/prices/data/"


We use Flume's file_roll type for this storage


This is an event driven architecture and our interest is to process these
trail files through SSS at a predetermined interval or when needed.
However, the caveat is that the volume should be containable meaning SSS
can process it as required. The code base in PySpark is as follows:


      data_path = "etcbucket/prices/data/"

      checkpoint_path = "etcbucket/prices/chkpt/"

        try:

            streamingDataFrame = self.spark \

                .readStream \

                .option('newFilesOnly', 'true') \

                .option('header', 'true') \

                .option('maxFilesPerTrigger', 10000) \

                .option('latestFirst', 'false') \

                .text(data_path) \

                .select(from_json(col("value").cast("string"),
schema).alias("parsed_value"))


            streamingDataFrame.printSchema()

            result = streamingDataFrame.select( \

                     col("parsed_value.rowkey").alias("rowkey") \

                   , col("parsed_value.ticker").alias("ticker") \

                   , col("parsed_value.timeissued").alias("timeissued") \

                   , col("parsed_value.price").alias("price")). \

                     writeStream. \

                     outputMode('append'). \

                     option("truncate", "false"). \

                     *foreachBatch(sendToSink). \*

                     queryName('trailFiles'). \

   *                  trigger(once = True). \*

  *                   option('checkpointLocation', checkpoint_path). \*

                     start(data_path)

        except Exception as e:

                print(f"""{e}, quitting""")

                sys.exit(1)

def sendToSink(df, batchId):
    if(len(df.take(1))) > 0:
        print(f"""batchId is {batchId}""")
        df.show(100,False)
        df. persist()
        # write to BigQuery batch table
        s.writeTableToBQ(df, "append",
config['MDVariables']['targetDataset'],config['MDVariables']['targetTable'])
        df.unpersist()
        print(f"""wrote to DB""")
    else:
        print("DataFrame is empty")



   view my Linkedin profile
<https://www.linkedin.com/in/mich-talebzadeh-ph-d-5205b2/>



*Disclaimer:* Use it at your own risk. Any and all responsibility for any
loss, damage or destruction of data or any other property which may arise
from relying on this email's technical content is explicitly disclaimed.
The author will in no case be liable for any monetary damages arising from
such loss, damage or destruction.




On Sun, 2 May 2021 at 10:42, Mail Delivery Subsystem <
mailer-daemon@googlemail.com> wrote:

> [image: Error Icon]
> Message too large
> Your message couldn't be delivered to *user@spark.apache.org* because it
> exceeds the size limit. Try reducing the message size and then resending
> it.
> The response from the remote server was:
>
> 552 5.3.4 Message size exceeds fixed limit
>
>
>
> ---------- Forwarded message ----------
> From: Mich Talebzadeh <mich.talebzadeh@gmail.com>
> To:
> Cc: user <user@spark.apache.org>
> Bcc:
> Date: Sun, 2 May 2021 10:42:09 +0100
> Subject: Re: Spark Streaming with Files
> ----- Message truncated -----

Mime
View raw message