Training models using streaming Kafka dataset

[dalelane]

I'm sure my problem is I'm using this wrong rather than that this is a bug.

I'm trying to train a model using a Kafka dataset, and continually refresh/retrain the model as new events appear on the topic.

A simplified version of my setup is:

batch_size=10

dataset = kafka_io.KafkaDataset([ "TRAINING.TOPIC:0" ],
                                servers="kafka.bootstrap:9092",
                                group="TRAINING.GROUP",
                                eof=False,
                                config_global=[
                                    "api.version.request=true",
                                    "enable.auto.commit=true"
                                ])

dataset = dataset.map(deserialize).batch(batch_size)

model = keras.Sequential([
    keras.layers.Flatten(input_shape=(28, 28)),
    keras.layers.Dense(128, activation="relu"),
    keras.layers.Dense(10, activation="softmax")
])

model.compile(optimizer="adam",
            loss="sparse_categorical_crossentropy",
            metrics=["accuracy"])

I'm using eof=False to keep the consumer going as new messages arrive, and then repeatedly calling model.fit to keep improving the model using the new data.

model.fit(dataset, epochs=1, steps_per_epoch=50)
model.fit(dataset, epochs=1, steps_per_epoch=50)
model.fit(dataset, epochs=1, steps_per_epoch=50)
...
model.fit(dataset, epochs=1, steps_per_epoch=50)

I was assuming it was safe to keep re-calling model.fit using the same open consumer dataset, but that assumption doesn't appear to be correct - it works okay for a little while before failing with a Too many open files error.

2020-02-21 08:12:17.026857: E tensorflow_io/kafka/kernels/kafka_dataset_ops.cc:156] 
EVENT_ERROR: (Local: Broker transport failure): sasl_ssl://9.20.193.37:31029/2: 
Failed to create socket: Too many open files (after 354960171ms in state INIT)

I guess it must be creating multiple new Kafka clients?

Can you help let me know what I'm doing wrong, please?

[yongtang]

@dalelane When a dataset is passed to model.fit, it is expected that dataset is repeatable so that at each epoch the same data (could be shuffled) is used again. In that case, the Kafka Dataset (or any other types of dataset passed to model.fit) will need to be both finite and repeatable. So EOF is necessary). The non-finite dataset (with eof=False) is mostly used for inference (model.predict) where each element in the dataset only needs to be used once.

The above is the normal setup commonly seen. Though by looking at your setup, it looks like you are trying to iteratively "retrain" the model with only the net-new data? There might be some way to make your pipeline work with some small adjustment, would you mind to explain a little further about your setup?

[dalelane]

Thanks for the reply! I don't have a strict requirement here - I'm more trying to understand what is possible using Kafka.

My initial hope was that I could treat each set of messages as a new batch - conceptually each new batch of Kafka events being part of one large never-ending epoch. I'd put a batch size of 10 just as a try-to-see-if-it-works rather than because I think that is a good value - but I was hoping that KafkaDataset would collect up to 10 messages, and once it has enough to make another new batch, I could do another iteration using that latest batch.

What I'm trying to understand is if a Kafka-based pipeline could be used to have a ML model that is constantly refined as new events arrive (rather than retrain from scratch with a complete copy of all events received until this point).

(I haven't tried whether there is a partial_fit or train_on_batch approach that might work with this - I've only tried with model.fit so far.)

Do you think what I'm describing is feasible?

[yongtang]

@dalelane I didn't work on retrain model before, though if I remember, it was possible to retrain the model with pure tensor input in the past. Whether or not the tf.data.Dataset works, we will have to find out.

You might need to save the model first (so that all the state information is finalized), then load the model from saved model file. After that, the model that was loaded could be used for continue training:

model.fit(inputs, ...)

model.save('my_model.h5')

model = keras.load_model('my_model.h5')

# continue fitting
model.fit(inputs,...)

With the recent changes in TF 2.0, I will have to validate to find out if the same still works or not.

Also for tf.data source, may need to check the tf.keras' implementation to find out as well.

[yongtang]

@dalelane By the way, tensorflow/io does have the lazy-load tensor format support for Kafka. So even if tf.data could not use used in model retrain, we could try with IOTensor instead.

[dalelane]

@yongtang Thanks for the reply.

I tried doing a save/load but unfortunately it didn't help.

(One thing I noticed when testing to see if it helped was that the EVENT_ERROR always occurs after the 73rd call to model.fit. Always immediately after the 73rd call. I hadn't realised it was so consistent in when it fails in my earlier testing)

I'll have a look at IOTensor - I've not tried that before, but I'll see how far I can get.

[yongtang]

@dalelane I am a little occupied but will try to take a look later this week. In the mean time, I would suggest you to take a look at the re-training with tensor created from simple numpy, to see if the idea of re-training with save/load model works first. If at least the retraining works on tensor created from simply numpy then we know the retraining idea works. We can definitely find out a way to make it in a streaming fashion like connecting to kafka. Otherwise I suspect some additional work from TensorFlow's main repo/tf.keras has to be done before we support the pipeline.

[dalelane]

@yongtang Sorry, I haven't had much luck.

I did try using IODataset - swapping out my dataset creation for this

dataset = IODataset.from_kafka("TRAINING.TOPIC",
    partition=0,
    offset=0,
    servers="kafka.bootstrap:9092",
    configuration=[
        "group.id=TRAINING.GROUP",
        "api.version.request=true",
        "enable.auto.commit=true"
    ])

On the plus side, repeated calls to model.fit worked without error'ing, so that was better.

But I'm not sure it wasn't just repeatedly using the same data, as the committed offset for the consumer group never increased.

FWIW - I did find this reference, which seems to support that repeated calls to model.fit should work, although that is a few years old now.

[yongtang]

@dalelane Sorry for the late reply. The tf.keras code base changed a lot with extra indirections to support multiple types of input during the big 1.x => 2.0 migration. I will need to spend extra time to see how it works now. I will try to take a look during the weekend.

[dalelane]

That’s very kind - thank you so much!

[kvignesh1420]

@dalelane any update/insight as to why the committed offset by the consumer-group in Kafka was not increasing? Is it because you are setting offset=0 while initializing the dataset and when model.fit is called in a while loop, it is using the same data (where offset=0 in Kafka topic) ? Just a thought.

[kvignesh1420]

@dalelane please take a look the updated kafka notebook which provides info on streaming-based inference and online learning. Closing as of now. Please reopen if required.

[dalelane]

@kvignesh1420 This looks fantastic, thanks so much!
Would it be okay if I submitted a small PR to tweak it to replace the Confluent-specific bits, so this could be used by any Apache Kafka user, please?

[kvignesh1420]

@dalelane that would be nice!