Styjun

As a GM, is it bad form to ask for a moment to think when improvising?

Which sphere is fastest?

Why is my arithmetic with a long long int behaving this way?

What to use instead of cling film to wrap pastry

When an imagined world resembles or has similarities with a famous world

Feasibility of lava beings?

What was the first story to feature the plot "the monsters were human all along"?

Why aren't nationalizations in Russia described as socialist?

Where are the "shires" in the UK?

How do LIGO and VIRGO know that a gravitational wave has its origin in a neutron star or a black hole?

Which US defense organization would respond to an invasion like this?

Copy previous line to current line from text file

Has the Hulk always been able to talk?

Is the book wrong about the Nyquist Sampling Criterion?

Are pressure-treated posts that have been submerged for a few days ruined?

What do "Sech" and "Vich" mean in this sentence?

Why is "breaking the mould" positively connoted?

Does XQuartz work on macOS Mojave?

It is as simple as ABC

Dangerous workplace travelling

Is Benjen dead?

Can I use a Cat5e cable with an RJ45 and Cat6 port?

Why do people keep telling me that I am a bad photographer?

Is 'contemporary' ambiguous and if so is there a better word?

how to run foreachRDD with arguments inside a for loop

How to merge two dictionaries in a single expression?How do I check if a list is empty?How do I check whether a file exists without exceptions?Is there a way to run Python on Android?How can I safely create a nested directory in Python?Accessing the index in 'for' loops?How do I sort a dictionary by value?“Least Astonishment” and the Mutable Default ArgumentHow do I list all files of a directory?Iterating over dictionaries using 'for' loops

.everyoneloves__top-leaderboard:empty,.everyoneloves__mid-leaderboard:empty,.everyoneloves__bot-mid-leaderboard:empty height:90px;width:728px;box-sizing:border-box;

0

I have a for loop creating kafka DStreams by iterating over a list of topic names. Then I want to apply a function to each RDD, but I need to pass the topic name into the function. What I am seeing is that inside the function only the last topic value is available (topic3). I understand that's because foreachRDD is lazily executed. Is there any way to pass in the topic name? I have to accomplish this without using structured streaming.

topics = ["topic1", "topic2", "topic3"]
ssc = StreamingContext(spark_context, 5)

def process_topic(rdd, topic_value):
 if not rdd.isEmpty():
 print "topic : "+topic_value
 df = rdd.toDF().show()

for topic in topics:
 directKafkaStream = KafkaUtils.createDirectStream(ssc, [topic],'metadata.broker.list': 'broker_url', 'auto.offset.reset': 'smallest')
 lines = directKafkaStream.map(lambda v: json.loads(v[1]))
 lines.foreachRDD(lambda x: process_topic(x, topic))

ssc.start()
time.sleep(10)
ssc.stop()

The output looks like this:

"topic3"

(df for topic1)

"topic3"

(df for topic2)

"topic3"

(df for topic3)

Would like to see:

"topic1"

(df for topic1)

"topic2"

(df for topic2)

"topic3"

(df for topic3)

python apache-kafka

share|improve this question

asked Mar 23 at 1:41

EVSEVS

11

add a comment | 

0

I have a for loop creating kafka DStreams by iterating over a list of topic names. Then I want to apply a function to each RDD, but I need to pass the topic name into the function. What I am seeing is that inside the function only the last topic value is available (topic3). I understand that's because foreachRDD is lazily executed. Is there any way to pass in the topic name? I have to accomplish this without using structured streaming.

topics = ["topic1", "topic2", "topic3"]
ssc = StreamingContext(spark_context, 5)

def process_topic(rdd, topic_value):
 if not rdd.isEmpty():
 print "topic : "+topic_value
 df = rdd.toDF().show()

for topic in topics:
 directKafkaStream = KafkaUtils.createDirectStream(ssc, [topic],'metadata.broker.list': 'broker_url', 'auto.offset.reset': 'smallest')
 lines = directKafkaStream.map(lambda v: json.loads(v[1]))
 lines.foreachRDD(lambda x: process_topic(x, topic))

ssc.start()
time.sleep(10)
ssc.stop()

The output looks like this:

"topic3"

(df for topic1)

"topic3"

(df for topic2)

"topic3"

(df for topic3)

Would like to see:

"topic1"

(df for topic1)

"topic2"

(df for topic2)

"topic3"

(df for topic3)

python apache-kafka

share|improve this question

asked Mar 23 at 1:41

EVSEVS

11

add a comment | 

I have a for loop creating kafka DStreams by iterating over a list of topic names. Then I want to apply a function to each RDD, but I need to pass the topic name into the function. What I am seeing is that inside the function only the last topic value is available (topic3). I understand that's because foreachRDD is lazily executed. Is there any way to pass in the topic name? I have to accomplish this without using structured streaming.

topics = ["topic1", "topic2", "topic3"]
ssc = StreamingContext(spark_context, 5)

def process_topic(rdd, topic_value):
 if not rdd.isEmpty():
 print "topic : "+topic_value
 df = rdd.toDF().show()

for topic in topics:
 directKafkaStream = KafkaUtils.createDirectStream(ssc, [topic],'metadata.broker.list': 'broker_url', 'auto.offset.reset': 'smallest')
 lines = directKafkaStream.map(lambda v: json.loads(v[1]))
 lines.foreachRDD(lambda x: process_topic(x, topic))

ssc.start()
time.sleep(10)
ssc.stop()

The output looks like this:

"topic3"

(df for topic1)

"topic3"

(df for topic2)

"topic3"

(df for topic3)

Would like to see:

"topic1"

(df for topic1)

"topic2"

(df for topic2)

"topic3"

(df for topic3)

python apache-kafka

share|improve this question

asked Mar 23 at 1:41

EVSEVS

11

topics = ["topic1", "topic2", "topic3"]
ssc = StreamingContext(spark_context, 5)

def process_topic(rdd, topic_value):
 if not rdd.isEmpty():
 print "topic : "+topic_value
 df = rdd.toDF().show()

for topic in topics:
 directKafkaStream = KafkaUtils.createDirectStream(ssc, [topic],'metadata.broker.list': 'broker_url', 'auto.offset.reset': 'smallest')
 lines = directKafkaStream.map(lambda v: json.loads(v[1]))
 lines.foreachRDD(lambda x: process_topic(x, topic))

ssc.start()
time.sleep(10)
ssc.stop()

share|improve this question

asked Mar 23 at 1:41

EVSEVS

11

share|improve this question

asked Mar 23 at 1:41

EVSEVS

11

asked Mar 23 at 1:41

EVSEVS

11

asked Mar 23 at 1:41

EVSEVS

11