Hi, Could anyone help me debug this? I have added all the jars I could find (being desperate here). The jar paths are correct because flink-parquet worked.

env = StreamExecutionEnvironment.get_execution_environment()
jars = "file:///mnt/c/Users/piby/Downloads/flink/examples/python/jars/flink-sql-parquet-1.16.1.jar; \
             file:///mnt/c/Users/piby/Downloads/flink/examples/python/jars/flink-parquet-1.16.1.jar; \
             file:///mnt/c/Users/piby/Downloads/flink/examples/python/jars/flink-hadoop-fs.jar; \
            file:///mnt/c/Users/piby/Downloads/flink/examples/python/jars/flink-hadoop-compatibility_2.12-1.16.1.jar; \
            file:///mnt/c/Users/piby/Downloads/flink/examples/python/jars/hadoop-common-3.3.4.jar; \
                file:///mnt/c/Users/piby/Downloads/flink/examples/python/jars/hadoop-client-3.3.4.jar"
env.add_jars(jars)
env.add_classpaths(jars)
env.set_runtime_mode(RuntimeExecutionMode.STREAMING)
# write all the data to one file
env.set_parallelism(1)

# FileSource

file_source = FileSource.for_record_stream_format(StreamFormat.text_line_format(), "/mnt/c/Users/piby/Downloads/flink/examples/python/input") \
                            .monitor_continuously(Duration.of_millis(5)) \
                            .build()
ds = env.from_source(file_source, watermark_strategy=WatermarkStrategy.no_watermarks(), source_name="File Source")
ds = ds.map(lambda i: (i, 1), output_type=Types.TUPLE([Types.STRING(), <http://Types.INT|Types.INT>()])) \
           .key_by(lambda i: i[0],  key_type=Types.STRING())\
           .reduce(lambda i, j: (i[0], i[1] + j[1]))

row_type = DataTypes.ROW([
    DataTypes.FIELD(name='word', data_type=DataTypes.STRING(), description="word name"),
    DataTypes.FIELD(name='count', data_type=<http://DataTypes.INT|DataTypes.INT>(), description="total count")
])


file_sink = FileSink.for_bulk_format(base_path=output_path, \
                writer_factory=ParquetBulkWriters.for_row_type(
                                                    row_type,
                                                    hadoop_config=Configuration(),
                                                    utc_timestamp=True,
                                                ))  \
                .with_bucket_assigner(BucketAssigner.date_time_bucket_assigner(format_str="yyyy/MM/dd/HH", timezone_id="UTC")) \
                .with_output_file_config( 
                    OutputFileConfig.builder() 
                    .with_part_prefix("prefix") 
                    .with_part_suffix(".parquet") 
                    .build()) \
                .with_rolling_policy(RollingPolicy.default_rolling_policy()) \
                .build()

ds.sink_to(sink=file_sink)
env.execute()

Here is the error stack

---------------------------------------------------------------------------
TypeError                                 Traceback (most recent call last)
Cell In[1], line 57
     46 ds = ds.map(lambda i: (i, 1), output_type=Types.TUPLE([Types.STRING(), <http://Types.INT|Types.INT>()])) \
     47            .key_by(lambda i: i[0],  key_type=Types.STRING())\
     48            .reduce(lambda i, j: (i[0], i[1] + j[1]))
     50 row_type = DataTypes.ROW([
     51     DataTypes.FIELD(name='word', data_type=DataTypes.STRING(), description="word name"),
     52     DataTypes.FIELD(name='count', data_type=<http://DataTypes.INT|DataTypes.INT>(), description="total count")
     53 ])
     56 file_sink = FileSink.for_bulk_format(base_path=output_path, \
---> 57                 writer_factory=ParquetBulkWriters.for_row_type(
     58                                                     row_type,
     59                                                     utc_timestamp=True,
     60                                                 ))  \
     61                 .with_bucket_assigner(BucketAssigner.date_time_bucket_assigner(format_str="yyyy/MM/dd/HH", timezone_id="UTC")) \
     62                 .with_output_file_config(
     63                     OutputFileConfig.builder()
     64                     .with_part_prefix("prefix")
     65                     .with_part_suffix(".parquet")
     66                     .build()) \
     67                 .with_rolling_policy(RollingPolicy.default_rolling_policy()) \
     68                 .build()
     70 ds.sink_to(sink=file_sink)
     71 env.execute()

File ~/anaconda3/envs/flink/lib/python3.8/site-packages/pyflink/datastream/formats/parquet.py:212, in ParquetBulkWriters.for_row_type(row_type, hadoop_config, utc_timestamp)
    208 jvm = get_gateway().jvm
    209 JParquetRowDataBuilder = jvm.org.apache.flink.formats.parquet.row.ParquetRowDataBuilder
    210 return RowDataBulkWriterFactory(JParquetRowDataBuilder.createWriterFactory(
    211     _to_java_data_type(row_type).getLogicalType(),
--> 212     create_hadoop_configuration(hadoop_config),
    213     utc_timestamp
    214 ), row_type)

File ~/anaconda3/envs/flink/lib/python3.8/site-packages/pyflink/datastream/utils.py:39, in create_hadoop_configuration(config)     
     37 def create_hadoop_configuration(config: Configuration):
     38     jvm = get_gateway().jvm
---> 39     hadoop_config = jvm.org.apache.hadoop.conf.Configuration()
     40     for k, v in config.to_dict().items():
     41         hadoop_config.set(k, v)

File ~/anaconda3/envs/flink/lib/python3.8/site-packages/pyflink/util/exceptions.py:185, in install_py4j_hooks.<locals>.wrapped_call(self, *args, **kwargs)
    184 def wrapped_call(self, *args, **kwargs):
--> 185     raise TypeError(
    186         "Could not found the Java class '%s'. The Java dependencies could be specified via "
    187         "command line argument '--jarfile' or the config option 'pipeline.jars'" % self._fqn)

TypeError: Could not found the Java class 'org.apache.hadoop.conf.Configuration'. The Java dependencies could be specified via command line argument '--jarfile' or the config option 'pipeline.jars'

Hi piby, parquet format needs some common hadoop libs, you can set env var HADOOP_CLASSPATH before executing pyflink job.