publicclassKafkaSinkSimpleTest01 {
publicstaticvoidmain(String[] args) throwsException {
// set up the streaming execution environmentfinalStreamExecutionEnvironmentenv=StreamExecutionEnvironment.getExecutionEnvironment();
// 简单添加几条string类型的数据DataStreamSource<String>stream=env.fromElements("赵", "钱", "孙", "李", "周");
// 做个print sink以便观察stream.print("print-test");
// 实现自定义的 KafkaSerializationSchemaKafkaSerializationSchema<String>serializationSchema=newKafkaSerializationSchema<String>() {
// 重写 serialize方法，并重新定义ProducerRecord，以指定数据的分区或者key@OverridepublicProducerRecord<byte[], byte[]>serialize(Stringelement, @NullableLongtimestamp) {
returnnewProducerRecord<byte[], byte[]>(
"sink_topic", // target topic11,  // partition"key-value".getBytes(StandardCharsets.UTF_8),  // keyelement.getBytes(StandardCharsets.UTF_8)); // record contents            }
        };
// 实例化 sinkfunctionFlinkKafkaProducer<String>myProducer=newFlinkKafkaProducer<>(
"sink_topic",             // target topicserializationSchema,    // serialization schemaproperties,             // producer configFlinkKafkaProducer.Semantic.EXACTLY_ONCE); // fault-tolerancestream.addSink(myProducer);
// execute programenv.execute("Flink Streaming Java API Skeleton");
    }
}

publicclassKafkaSinkWithSelfDefPartitioner {
publicstaticvoidmain(String[] args) throwsException {
// set up the streaming execution environmentfinalStreamExecutionEnvironmentenv=StreamExecutionEnvironment.getExecutionEnvironment();
// 简单添加几条string类型的数据DataStreamSource<String>stream=env.fromElements("赵", "钱", "孙", "李", "周");
stream.print("print-test");
// 在SerializationSchema的实现中不指定分区SerializationSchema<String>serializationSchema=newSerializationSchema<String>(
        ) {
@Overridepublicbyte[] serialize(Stringelement) {
returnelement.getBytes(StandardCharsets.UTF_8);
            }
        };
// 自定义FlinkKafkaPartitioner 根据record的值进行判断，写入不同的分区FlinkKafkaPartitioner<String>flinkKafkaPartitioner=newFlinkKafkaPartitioner<String>() {
@Overridepublicintpartition(Stringrecord, byte[] key, byte[] value, StringtargetTopic, int[] partitions) {
// 主要是根据record的情况，更改这里的实现方法，返回kafka分区的integer类型即可if(record.equals("test")){
return6;
                }else{
return7;
                }
            }
        };
FlinkKafkaProducer<String>myProducer=newFlinkKafkaProducer<>(
"sunyf_topic",
serializationSchema,
properties,
flinkKafkaPartitioner,
FlinkKafkaProducer.Semantic.EXACTLY_ONCE,
2        );
stream.addSink(myProducer);
// execute programenv.execute("Flink Streaming Java API Skeleton");
    }
}

packageorg.example;
importorg.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
importorg.apache.flink.table.data.binary.BinaryRowData;
publicclassFlinkSQLKafkaSinkPartitionerextendsFlinkKafkaPartitioner {
@Overridepublicintpartition(Objectrecord, byte[] key, byte[] value, StringtargetTopic, int[] partitions) {
BinaryRowDatabrData=null;
// 在flink sql中配置自定义partitioner record的实际类型是BinaryRowDataif (recordinstanceofBinaryRowData) {
brData= (BinaryRowData) record;
// 可以通过如下类的方法获取该行所有数据for (inti=0; i<brData.getArity(); i++) {
System.out.println("索引: "+i+" :"+brData.getString(i));
            }
Stringdt=brData.getString(2).toString();
intdtInt=Integer.parseInt(dt);
//        单数日期写5分区，偶数写4分区if (dtInt%2==0){
return4;
            }else{
return5;
            }
        } else {
thrownewIllegalArgumentException("Record is not of type BinaryRowData");
        }
    }
}

Propertiesproperties=newProperties();
properties.setProperty("bootstrap.servers", "ip:9093,ip:9093,ip:9093");
properties.setProperty("security.protocol", "SASL_SSL");
// SSL 配置// 配置服务端提供的 truststore (CA 证书) 的路径// Configure the path of truststore (CA) provided by the serverproperties.setProperty("ssl.truststore.location", "/file_path/mix.4096.client.truststore.jks");
properties.setProperty("ssl.truststore.password", "KafkaOnsClient");
// SASL 配置// 将 SASL 机制配置为 as SCRAM-SHA-256properties.setProperty("sasl.mechanism", "PLAIN");
properties.setProperty("ssl.endpoint.identification.algorithm", "");
// 配置 JAASproperties.setProperty("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"username\" password=\"password\";");