0

After read official docs, i tried using checkpoint with getOrCreate in spark streaming. Some snippets:

def get_ssc():
    sc = SparkContext("yarn-client")
    ssc = StreamingContext(sc, 10)  # calc every 10s
    ks = KafkaUtils.createDirectStream(
        ssc, ['lucky-track'], {"metadata.broker.list": KAFKA_BROKER})
    process_data(ks)

    ssc.checkpoint(CHECKPOINT_DIR)
    return ssc

if __name__ == '__main__':
    ssc = StreamingContext.getOrCreate(CHECKPOINT_DIR, get_ssc)

    ssc.start()
    ssc.awaitTermination()

The code works fine for recover, but the recovered context always works on the old process function. It means that even if i changed map/reduce function code, it not works at all.

Until now, spark(1.5.2) still not support arbitrary offset for python. So, what should i do to make this work properly?

zero323
  • 322,348
  • 103
  • 959
  • 935
Terran
  • 649
  • 10
  • 24

1 Answers1

0

Such behaviour is "by design", and is valid also for java/scala Spark applications. Entire code is serialized while checkpointing. If code changes, checkpoint data should be truncated.

Alex Krash
  • 1
  • umm...I've got this. So, any idea about the right way to recover except recording offset by myself? – Terran Dec 24 '15 at 01:52