class_docstr: Convenience builder to create :class:`~pyflink.common.serialization.BulkWriterFactory` for
  Avro types.
  
  .. versionadded:: 1.16.0
class_docstr_tok: ["Convenience","builder","to","create",":","class",":","`","~pyflink.common.serialization.BulkWriterFactory","`","for","Avro","types",".","..","versionadded",":",":","1.16.0"]
docstr: Creates an AvroWriterFactory that accepts and writes Avro generic types. The Avro writers
  will use the given schema to build and write the records.
  
  Note that to make this works in PyFlink, you need to declare the output type of the
  predecessor before FileSink to be :class:`~GenericRecordAvroTypeInfo`, and the predecessor
  cannot be :meth:`StreamExecutionEnvironment.from_collection`, you can add a pass-through map
  function before the sink, as the example shown below.
  
  The Python data records should match the Avro schema, and have the same behavior with
  vanilla Python data structure, e.g. an object for Avro array should behave like Python list,
  an object for Avro map should behave like Python dict.
  
  Example:
  ::
  
      >>> env = StreamExecutionEnvironment.get_execution_environment()
      >>> schema = AvroSchema(JSON_SCHEMA)
      >>> avro_type_info = GenericRecordAvroTypeInfo(schema)
      >>> ds = env.from_collection([{'array': [1, 2]}], type_info=Types.PICKLED_BYTE_ARRAY())
      >>> sink = FileSink.for_bulk_format(
      ...     OUTPUT_DIR, AvroBulkWriters.for_generic_record(schema)).build()
      >>> # A map to indicate its Avro type info is necessary for serialization
      >>> ds.map(lambda e: e, output_type=GenericRecordAvroTypeInfo(schema)) \
      ...     .sink_to(sink)
  
  :param schema: The avro schema.
  :return: The BulkWriterFactory to write generic records into avro files.
docstr_tok: ["Creates","an","AvroWriterFactory","that","accepts","and","writes","Avro","generic","types",".","The","Avro","writers","will","use","the","given","schema","to","build","and","write","the","records",".","Note","that","to","make","this","works","in","PyFlink",",","you","need","to","declare","the","output","type","of","the","predecessor","before","FileSink","to","be",":","class",":","`","~GenericRecordAvroTypeInfo","`",",","and","the","predecessor","can","not","be",":","meth",":","`","StreamExecutionEnvironment.from_collection","`",",","you","can","add","a","pass-through","map","function","before","the","sink",",","as","the","example","shown","below",".","The","Python","data","records","should","match","the","Avro","schema",",","and","have","the","same","behavior","with","vanilla","Python","data","structure",",","e.g",".","an","object","for","Avro","array","should","behave","like","Python","list",",","an","object","for","Avro","map","should","behave","like","Python","dict",".","Example",":",":",":",">",">",">","env","=","StreamExecutionEnvironment.get_execution_environment","(",")",">",">",">","schema","=","AvroSchema","(","JSON_SCHEMA",")",">",">",">","avro_type_info","=","GenericRecordAvroTypeInfo","(","schema",")",">",">",">","ds","=","env.from_collection","(","[","{","'array","'",":","[","1",",","2","]","}","]",",","type_info=Types.PICKLED_BYTE_ARRAY","(",")",")",">",">",">","sink","=","FileSink.for_bulk_format","(","...","OUTPUT_DIR",",","AvroBulkWriters.for_generic_record","(","schema",")",")",".build","(",")",">",">",">","#","A","map","to","indicate","its","Avro","type","info","is","necessary","for","serialization",">",">",">","ds.map","(","lambda","e",":","e",",","output_type=GenericRecordAvroTypeInfo","(","schema",")",")","\\","...",".sink_to","(","sink",")",":","param","schema",":","The","avro","schema",".",":","return",":","The","BulkWriterFactory","to","write","generic","records","into","avro","files","."]
code: def for_generic_record(schema: "AvroSchema") -> "BulkWriterFactory":
      """
          Creates an AvroWriterFactory that accepts and writes Avro generic types. The Avro writers
          will use the given schema to build and write the records.
  
          Note that to make this works in PyFlink, you need to declare the output type of the
          predecessor before FileSink to be :class:`~GenericRecordAvroTypeInfo`, and the predecessor
          cannot be :meth:`StreamExecutionEnvironment.from_collection`, you can add a pass-through map
          function before the sink, as the example shown below.
  
          The Python data records should match the Avro schema, and have the same behavior with
          vanilla Python data structure, e.g. an object for Avro array should behave like Python list,
          an object for Avro map should behave like Python dict.
  
          Example:
          ::
  
              >>> env = StreamExecutionEnvironment.get_execution_environment()
              >>> schema = AvroSchema(JSON_SCHEMA)
              >>> avro_type_info = GenericRecordAvroTypeInfo(schema)
              >>> ds = env.from_collection([{'array': [1, 2]}], type_info=Types.PICKLED_BYTE_ARRAY())
              >>> sink = FileSink.for_bulk_format(
              ...     OUTPUT_DIR, AvroBulkWriters.for_generic_record(schema)).build()
              >>> # A map to indicate its Avro type info is necessary for serialization
              >>> ds.map(lambda e: e, output_type=GenericRecordAvroTypeInfo(schema)) \\
              ...     .sink_to(sink)
  
          :param schema: The avro schema.
          :return: The BulkWriterFactory to write generic records into avro files.
          """
      jvm = get_gateway().jvm
      j_bulk_writer_factory = jvm.org.apache.flink.formats.avro.AvroWriters.forGenericRecord(
          schema._j_schema
      )
      return BulkWriterFactory(j_bulk_writer_factory)
  
imports: from py4j.java_gateway import get_java_class, JavaObject, java_import
  from pyflink.common.io import InputFormat
  from pyflink.common.serialization import BulkWriterFactory, SerializationSchema, DeserializationSchema
  from pyflink.common.typeinfo import TypeInformation
  from pyflink.datastream.utils import ResultTypeQueryable
  from pyflink.java_gateway import get_gateway
  from pyflink.util.java_utils import get_field_value, load_java_class
label_desc: Use image node_id 1 for calling the AvroBulkWriters obj's underlying member method code with example usage: obj.for_generic_record(schema) and returns: BulkWriterFactory