Let us look at your kafka streams.
Say we just read them like below
first read data from the topic
schema = StructType().add("rowkey", StringType()).add("ticker",
StringType()).add("timeissued", TimestampType()).add("price", FloatType())
try:
# construct a streaming dataframe streamingDataFrame that
subscribes to topic config['MDVariables']['topic']) -> md (market data)
streamingDataFrame = self.spark \
.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers",
config['MDVariables']['bootstrapServers'],) \
.option("schema.registry.url",
config['MDVariables']['schemaRegistryURL']) \
.option("group.id", config['common']['appName']) \
.option("zookeeper.connection.timeout.ms",
config['MDVariables']['zookeeperConnectionTimeoutMs']) \
.option("rebalance.backoff.ms",
config['MDVariables']['rebalanceBackoffMS']) \
.option("zookeeper.session.timeout.ms",
config['MDVariables']['zookeeperSessionTimeOutMs']) \
.option("auto.commit.interval.ms",
config['MDVariables']['autoCommitIntervalMS']) \
.option("subscribe", config['MDVariables']['topic']) \
.option("failOnDataLoss", "false") \
.option("includeHeaders", "true") \
.option("startingOffsets", "latest") \
.load() \
.select(from_json(col("value").cast("string"),
schema).alias("parsed_value"))
# then do writestream
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(SendToBigQuery). \
trigger(processingTime='2 seconds'). \
start()
except Exception as e:
print(f"""{e}, quitting""")
sys.exit(1)
result.awaitTermination()
Now you have microbatch every 2 seconds that calls method SendToBigQuery()
and that is what you need for your work
def SendToBigQuery(df, batchId): ## this is where you are receiving your
message read in dataframe DF and the batchId
if(len(df.take(1))) > 0:
#df.printSchema()
df. persist()
spark_session = s.spark_session(config['common']['appName'])
spark_session = s.setSparkConfBQ(spark_session)
# read from BigQuery that is your reference data that you are
reading from your RDBMS table
read_df = s.loadTableFromBQ(spark_session,
config['MDVariables']['targetDataset'],config['MDVariables']['targetTable'])
# look for high value tickers This is the one where you need to
go through every row of streaming dataframe and get your device_id (rowkey
is your device_id here)
*for row in df.rdd.collect():*
rowkey = row.rowkey
ticker = row.ticker
price = row.price
values = bigQueryAverages(ticker,price,read_df)
Average = values["average"]
standardDeviation = values["standardDeviation"]
lower = values["lower"]
upper = values["upper"]
if lower is not None and upper is not None:
hvTicker = priceComparison(ticker,price,lower,upper)
if(hvTicker == 1):
writeHighValueData(df,rowkey)
df.unpersist()
else:
print("DataFrame is empty")
HTH
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 Mon, 3 May 2021 at 18:01, Eric Beabes <mailinglist...@gmail.com> wrote:
> 1) Device_id might be different for messages in a batch.
> 2) It's a Streaming application. The IOT messages are getting read in a
> Structured Streaming job in a "Stream". The Dataframe would need to be
> updated every hour. Have you done something similar in the past? Do you
> have an example to share?
>
> On Mon, May 3, 2021 at 9:52 AM Mich Talebzadeh <mich.talebza...@gmail.com>
> wrote:
>
>> Can you please clarify:
>>
>>
>> 1. The IOT messages in one batch have the same device_id or every row
>> has different device_id?
>> 2. The RDBMS table can be read through JDBC in Spark and a dataframe
>> can be created on. Does that work for you? You do not really need to
>> stream
>> the reference table.
>>
>>
>> HTH
>>
>>
>>
>> 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 Mon, 3 May 2021 at 17:37, Eric Beabes <mailinglist...@gmail.com>
>> wrote:
>>
>>> I would like to develop a Spark Structured Streaming job that reads
>>> messages in a Stream which needs to be “joined” with another Stream of
>>> “Reference” data.
>>>
>>> For example, let’s say I’m reading messages from Kafka coming in from
>>> (lots of) IOT devices. This message has a ‘device_id’. We have a DEVICE
>>> table on a relational database. What I need to do is “join” the ‘device_id’
>>> in the message with the ‘device_id’ on the table to enrich the incoming
>>> message. Somewhere I read that, this can be done by joining two streams. I
>>> guess, we can create a “Stream” that reads the DEVICE table once every hour
>>> or so.
>>>
>>> Questions:
>>> 1) Is this the right way to solve this use case?
>>> 2) Should we use a Stateful Stream for reading DEVICE table with State
>>> timeout set to an hour?
>>> 3) What would happen while the DEVICE state is getting updated from the
>>> table on the relational database?
>>>
>>> Guidance would be greatly appreciated. Thanks.
>>>
>>
