docstr: Read messages from the underlying KafkaDatasource connection and create
  a cudf Dataframe
  
  Parameters
  ----------
  topic : str,
      Name of the Kafka topic that the messages
      should be read from
  partition : int,
      Partition number on the specified topic that
      should be read from
  lines : {{ True, False }}, default True,
      Whether messages should be treated as individual lines
  start : int, default 0,
      The beginning offset that should be used when
      reading a range of messages
  end : int, default 0,
      The last offset that will be read when
      reading a range of messages
  batch_timeout : int, default 10000,
      Amount of time to wait on the
      reading of the messages from Kafka in Milliseconds
  delimiter : str, default "\n",
      If lines=True this is the delimiter that
      will be placed between all messages that are read from Kafka
  message_format : {{ 'avro', 'csv', 'json', 'orc', 'parquet' }},
  default 'json',
      Format of the messages that will be read from Kafka.
      This dictates which underlying cudf reader will be invoked the
      create the Dataframe.
  
  Returns
  -------
  DataFrame
docstr_tok: ["Read","messages","from","the","underlying","KafkaDatasource","connection","and","create","a","cudf","Dataframe","Parameters","--","--","--","--","--","topic",":","str",",","Name","of","the","Kafka","topic","that","the","messages","should","be","read","from","partition",":","int",",","Partition","number","on","the","specified","topic","that","should","be","read","from","lines",":","{","{","True",",","False","}","}",",","default","True",",","Whether","messages","should","be","treated","as","individual","lines","start",":","int",",","default","0",",","The","beginning","offset","that","should","be","used","when","reading","a","range","of","messages","end",":","int",",","default","0",",","The","last","offset","that","will","be","read","when","reading","a","range","of","messages","batch_timeout",":","int",",","default","10000",",","Amount","of","time","to","wait","on","the","reading","of","the","messages","from","Kafka","in","Milliseconds","delimiter",":","str",",","default","``","\\n","''",",","If","lines=True","this","is","the","delimiter","that","will","be","placed","between","all","messages","that","are","read","from","Kafka","message_format",":","{","{","'avro","'",",","'csv","'",",","'json","'",",","'orc","'",",","'parquet","'","}","}",",","default","'json","'",",","Format","of","the","messages","that","will","be","read","from","Kafka",".","This","dictates","which","underlying","cudf","reader","will","be","invoked","the","create","the","Dataframe",".","Returns","--","--","--","-","DataFrame"]
code: def read_gdf(
      self,
      topic=None,
      partition=0,
      lines=True,
      start=0,
      end=0,
      batch_timeout=10000,
      delimiter="\n",
      message_format="json",
  ):
      r"""
      Read messages from the underlying KafkaDatasource connection and create
      a cudf Dataframe
  
      Parameters
      ----------
      topic : str,
          Name of the Kafka topic that the messages
          should be read from
      partition : int,
          Partition number on the specified topic that
          should be read from
      lines : {{ True, False }}, default True,
          Whether messages should be treated as individual lines
      start : int, default 0,
          The beginning offset that should be used when
          reading a range of messages
      end : int, default 0,
          The last offset that will be read when
          reading a range of messages
      batch_timeout : int, default 10000,
          Amount of time to wait on the
          reading of the messages from Kafka in Milliseconds
      delimiter : str, default "\n",
          If lines=True this is the delimiter that
          will be placed between all messages that are read from Kafka
      message_format : {{ 'avro', 'csv', 'json', 'orc', 'parquet' }},
      default 'json',
          Format of the messages that will be read from Kafka.
          This dictates which underlying cudf reader will be invoked the
          create the Dataframe.
  
      Returns
      -------
      DataFrame
      """
  
      if topic is None:
          raise ValueError(
              "ERROR: You must specify the topic "
              "that you want to consume from"
          )
  
      kafka_datasource = KafkaDatasource(
          self.kafka_configs,
          topic.encode(),
          partition,
          start,
          end,
          batch_timeout,
          delimiter.encode(),
      )
  
      cudf_readers = {
          "json": cudf.io.read_json,
          "csv": cudf.io.read_csv,
          "orc": cudf.io.read_orc,
          "avro": cudf.io.read_avro,
          "parquet": cudf.io.read_parquet,
      }
  
      result = cudf_readers[message_format](
          kafka_datasource, engine="cudf", lines=True
      )
  
      # Close up the cudf datasource instance
      # TODO: Ideally the C++ destructor should handle the
      # unsubscribe and closing the socket connection.
      kafka_datasource.unsubscribe()
      kafka_datasource.close(batch_timeout)
  
      if result is not None:
          if isinstance(result, cudf.DataFrame):
              return result
          else:
              return cudf.DataFrame._from_data(result)
      else:
          # empty Dataframe
          return cudf.DataFrame()
  
file_path: cudf/python/custreamz/custreamz/kafka.py
imports: import confluent_kafka
  from cudf_kafka._lib.kafka import KafkaDatasource
  import cudf
label_desc: Use image node_id 2 for calling the Consumer obj's underlying member method code with example usage: obj.read_gdf(topic, partition, lines, start, end, batch_timeout, delimiter, message_format) and returns: cudf, result, cudf