import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer010; //导入依赖的package包/类
private static void writeEnrichedStream(DataStream<AisMessage> enrichedAisMessagesStream,
    String parsingConfig, boolean writeOutputStreamToFile, String outputLineDelimiter,
    String outputPath, String outputStreamTopic) throws IOException {

  if (writeOutputStreamToFile) {
    enrichedAisMessagesStream.map(new AisMessagesToCsvMapper(outputLineDelimiter)).writeAsText(
        outputPath, WriteMode.OVERWRITE);
    return;
  }

  // Write to Kafka
  Properties producerProps = AppUtils.getKafkaProducerProperties();

  FlinkKafkaProducer010Configuration<AisMessage> myProducerConfig =
      FlinkKafkaProducer010.writeToKafkaWithTimestamps(enrichedAisMessagesStream,
          outputStreamTopic, new AisMessageCsvSchema(parsingConfig, outputLineDelimiter),
          producerProps);
  myProducerConfig.setLogFailuresOnly(false);
  myProducerConfig.setFlushOnCheckpoint(true);

}
 

import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer010; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
    // Read parameters from command line
    final ParameterTool params = ParameterTool.fromArgs(args);

    if(params.getNumberOfParameters() < 4) {
        System.out.println("\nUsage: FlinkReadKafka --read-topic <topic> --write-topic <topic> --bootstrap.servers <kafka brokers> --group.id <groupid>");
        return;
    }


    // setup streaming environment
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.getConfig().setRestartStrategy(RestartStrategies.fixedDelayRestart(4, 10000));
    env.enableCheckpointing(300000); // 300 seconds
    env.getConfig().setGlobalJobParameters(params);

    DataStream<String> messageStream = env
            .addSource(new FlinkKafkaConsumer010<>(
                    params.getRequired("read-topic"),
                    new SimpleStringSchema(),
                    params.getProperties())).name("Read from Kafka");

    // setup table environment
    StreamTableEnvironment sTableEnv = TableEnvironment.getTableEnvironment(env);


    // Write JSON payload back to Kafka topic
    messageStream.addSink(new FlinkKafkaProducer010<>(
                params.getRequired("write-topic"),
                new SimpleStringSchema(),
                params.getProperties())).name("Write To Kafka");

    env.execute("FlinkReadWriteKafka");
}
 

import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer010; //导入依赖的package包/类
public static void main(String[] args) throws Exception {

		TaxiRideCleansingParameterParser params = new TaxiRideCleansingParameterParser();
		// TODO: refactor this method
		if(!params.parseParams(args)){
			final String dataFilePath = params.getDataFilePath();

			// get an ExecutionEnvironment
			StreamExecutionEnvironment env =
					StreamExecutionEnvironment.getExecutionEnvironment();
			// configure event-time processing
			env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

			// get the taxi ride data stream
			DataStream<TaxiRide> rides = env.addSource(
					new TaxiRideSource(dataFilePath, MAX_EVENT_DELAY_DEFAULT, SERVING_SPEED_FACTOR_DEFAULT));

			TaxiRideCleansing taxiRideCleansing = new TaxiRideCleansing();
			DataStream<TaxiRide> filteredRides = taxiRideCleansing.execute(rides);

			filteredRides.addSink(new FlinkKafkaProducer010<>(
					"localhost:9092",      // Kafka broker host:port
					"cleansedRides",       // Topic to write to
					new TaxiRideSchema())  // Serializer (provided as util)
			);

//			filteredRides.print();
			env.execute("Running Taxi Ride Cleansing");
		}
	}
 

import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer010; //导入依赖的package包/类
public static void configuration(DataStream<String> stream, String topic, Properties properties) {

        // using Apache Kafka as a sink for serialized generic output
        FlinkKafkaProducer010.FlinkKafkaProducer010Configuration kafkaConfig = FlinkKafkaProducer010
                .writeToKafkaWithTimestamps(
                        stream,
                        topic,
                        new SimpleStringSchema(),
                        properties
        );
        kafkaConfig.setLogFailuresOnly(false);
        kafkaConfig.setFlushOnCheckpoint(true);
    }
 

import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer010; //导入依赖的package包/类
public static void configuration(DataStream<StreetLamp> stream, Properties properties) {

        // using Apache Kafka as
        FlinkKafkaProducer010.FlinkKafkaProducer010Configuration kafkaConfig = FlinkKafkaProducer010
                .writeToKafkaWithTimestamps(
                        stream,
                        "control",
                        new ControlSerializationSchema(),
                        properties
                );

        kafkaConfig.setLogFailuresOnly(false);
        kafkaConfig.setFlushOnCheckpoint(true);
    }
 

import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer010; //导入依赖的package包/类
/**
 * The main entry method
 * 
 */
public static void main(String[] args) throws Exception {

  String cehkPointsPath =
      Paths.get(configs.getStringProp("flinkCheckPointsPath") + "/" + System.currentTimeMillis())
          .toUri().toString();


  int parallelism = configs.getIntProp("parallelism");
  String inputHdfsFile = configs.getStringProp("inputHDFSFilePath");
  String outputTopicName = configs.getStringProp("outputHDFSKafkaTopic");

  // Set up the execution environment
  final StreamExecutionEnvironment env =
      new StreamExecutionEnvBuilder().setParallelism(parallelism).setStateBackend(cehkPointsPath)
          .build();
  // Read the HDFS file
  DataStreamSource<String> inputTextStream =
      env.readTextFile(inputHdfsFile).setParallelism(parallelism);

  FlinkKafkaProducer010Configuration<String> myProducerConfig =
      FlinkKafkaProducer010.writeToKafkaWithTimestamps(inputTextStream, outputTopicName,
          new SimpleStringSchema(), AppUtils.getKafkaProducerProperties());


  myProducerConfig.setLogFailuresOnly(false);
  myProducerConfig.setFlushOnCheckpoint(true);


  System.out.println(env.getExecutionPlan());

  JobExecutionResult executionResult = null;

  try {
    executionResult = env.execute(" HDFS to Kafka stream producer");
  } catch (Exception e) {
    System.out.println(e.getMessage());
  }

  System.out.println("Full execution time=" + executionResult.getNetRuntime(TimeUnit.MINUTES));
}
 

import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer010; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
	// parse input arguments
	final ParameterTool parameterTool = ParameterTool.fromArgs(args);

	if (parameterTool.getNumberOfParameters() < 5) {
		System.out.println("Missing parameters!\n" +
				"Usage: Kafka --input-topic <topic> --output-topic <topic> " +
				"--bootstrap.servers <kafka brokers> " +
				"--zookeeper.connect <zk quorum> --group.id <some id> [--prefix <prefix>]");
		return;
	}

	String prefix = parameterTool.get("prefix", "PREFIX:");

	StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
	env.getConfig().disableSysoutLogging();
	env.getConfig().setRestartStrategy(RestartStrategies.fixedDelayRestart(4, 10000));
	env.enableCheckpointing(5000); // create a checkpoint every 5 seconds
	env.getConfig().setGlobalJobParameters(parameterTool); // make parameters available in the web interface

	// make parameters available in the web interface
	env.getConfig().setGlobalJobParameters(parameterTool);

	DataStream<String> input = env
			.addSource(new FlinkKafkaConsumer010<>(
					parameterTool.getRequired("input-topic"),
					new SimpleStringSchema(),
					parameterTool.getProperties()))
			.map(new PrefixingMapper(prefix));

	input.addSink(
			new FlinkKafkaProducer010<>(
					parameterTool.getRequired("output-topic"),
					new SimpleStringSchema(),
					parameterTool.getProperties()));

	env.execute("Kafka 0.10 Example");
}