Hello, I'm following Flink user guides to try to setup local environment where I can test very simple case of streaming data in parquet format into S3 with python runtime. I use bitnami helm charts for minio and flink, and as a client I use the simple pyflink repl found from the apache-flink 1.16.1 pypi package. My first attempt was to test with csv, but bumped into this issue https://issues.apache.org/jira/browse/FLINK-28513 . Next idea was to tryout parquet format as output (as it is my ultimate goal as dataformat in my solution), but started to think that where should I define the jar includes the

parquet

formatFactory. So, I'm keen to know that: 1. Which jar should I use (I'm using TableDescriptor to define my sink table)? flink-parquet or flink-sql-parquet? 2. Should I include it into cluster job and taskmanagers /opt/flink/lib/ folder or try add is as

.add_jar(...)

method? pyflink-shell doesn't offer --jars option during launch

You need to set a jar path in Python and make sure that you have the right jar that probably would not be included in a vanilla Flink install:

t_env.get_config().set("pipeline.jars", jar_path)

Maybe test it stand alone from your laptop before messing with Helm and friends. You might need to set your AWS settings. But possibly from inside your test program.

Hi Sami 🙂 ! Thanks for the tip, I'll try that one as well at some point. I do have full control of the Flink cluster (minikube, running localhost) so friends are safe 😄 . Reason for this is that local S3 (minio) configuration can only done with flink-conf.yaml configuration, due this https://issues.apache.org/jira/browse/FLINK-19589 I have deducted. I got the pipeline working when I added following jars into cluster nodes lib, i.e. classpath: flink-s3-fs-hadoop-1.16.1.jar, flink-parquet-1.16.1.jar, flink-sql-parquet-1.16.0.jar, hadoop-mapreduce-client-core-3.3.4.jar . Client python code needed these jars, flink-sql-parquet-1.16.0.jar and flink-s3-fs-hadoop-1.16.0.jar

Nice to see it works! Regarding flink-parquet vs flink-sql-parquet, you should use flink-sql-parquet as it’s a fat JAR which bundles some transitive dependencies in it.

So client can upload application specific jars to cluster to be in use and does it automatically when I add them via add_jars or setting "pipeline.jars"?

They are uploaded to the cluster automatically as part of the work of job submit. (Not uploaded when add_jars are called, upload happens during job submit)