Create a KafkaReader. Args: topics: A `tf.string` tensor containing one or more subscriptions, in the format of [topic:partition:offset:length], by default length is -1 for unlimited. servers: A list of bootstrap servers. group: The consumer group i
(
self,
topics,
servers="localhost",
group="",
eof=False,
timeout=1000,
config_global=None,
config_topic=None,
message_key=False,
)
| 502 | """ |
| 503 | |
| 504 | def __init__( |
| 505 | self, |
| 506 | topics, |
| 507 | servers="localhost", |
| 508 | group="", |
| 509 | eof=False, |
| 510 | timeout=1000, |
| 511 | config_global=None, |
| 512 | config_topic=None, |
| 513 | message_key=False, |
| 514 | ): |
| 515 | """Create a KafkaReader. |
| 516 | |
| 517 | Args: |
| 518 | topics: A `tf.string` tensor containing one or more subscriptions, |
| 519 | in the format of [topic:partition:offset:length], |
| 520 | by default length is -1 for unlimited. |
| 521 | servers: A list of bootstrap servers. |
| 522 | group: The consumer group id. |
| 523 | eof: If True, the kafka reader will stop on EOF. |
| 524 | timeout: The timeout value for the Kafka Consumer to wait |
| 525 | (in millisecond). |
| 526 | config_global: A `tf.string` tensor containing global configuration |
| 527 | properties in [Key=Value] format, |
| 528 | eg. ["enable.auto.commit=false", |
| 529 | "heartbeat.interval.ms=2000"], |
| 530 | please refer to 'Global configuration properties' |
| 531 | in librdkafka doc. |
| 532 | config_topic: A `tf.string` tensor containing topic configuration |
| 533 | properties in [Key=Value] format, |
| 534 | eg. ["auto.offset.reset=earliest"], |
| 535 | please refer to 'Topic configuration properties' |
| 536 | in librdkafka doc. |
| 537 | message_key: If True, the kafka will output both message value and key. |
| 538 | """ |
| 539 | self._topics = ops.convert_to_tensor(topics, dtype=dtypes.string, name="topics") |
| 540 | self._servers = ops.convert_to_tensor( |
| 541 | servers, dtype=dtypes.string, name="servers" |
| 542 | ) |
| 543 | self._group = ops.convert_to_tensor(group, dtype=dtypes.string, name="group") |
| 544 | self._eof = ops.convert_to_tensor(eof, dtype=dtypes.bool, name="eof") |
| 545 | self._timeout = ops.convert_to_tensor( |
| 546 | timeout, dtype=dtypes.int64, name="timeout" |
| 547 | ) |
| 548 | config_global = config_global if config_global else [] |
| 549 | self._config_global = ops.convert_to_tensor( |
| 550 | config_global, dtype=dtypes.string, name="config_global" |
| 551 | ) |
| 552 | config_topic = config_topic if config_topic else [] |
| 553 | self._config_topic = ops.convert_to_tensor( |
| 554 | config_topic, dtype=dtypes.string, name="config_topic" |
| 555 | ) |
| 556 | self._message_key = message_key |
| 557 | super(KafkaDataset, self).__init__() |
| 558 | |
| 559 | def _inputs(self): |
| 560 | return [] |