大数据处理框架：Flink：Flink端到端实时数据处理.docx

import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

public class KafkaDataStreamExample {

public static void main(String[] args) throws Exception {

//

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnviron

ment();

//

FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(

"input-topic", //

new SimpleStringSchema(), //

@Override

public String map(String value) throws Exception {

return value.toUpperCase(); //

}

});

//

env.execute("Kafka Data Stream Example");

}

}

public class CSVBatchExample {

public static void main(String[] args) throws Exception {

//

final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

//

DataSet<String> data = env.readTextFile("path/to/csv");

//

DataSet<Tuple2<String, Integer>> result = data.map(new MapFunction<String, Tuple2<Strin

g, Integer>>() {

@Override

result.print();

}

}

export FLINK_HOME=/path/to/your/flink-1.14.0

export PATH=$PATH:$FLINK_HOME/bin

source ~/.bashrc

source ~/.bash_profile

cd $FLINK_HOME/examples

$FLINK_HOME/bin/flink run -m yarn-cluster -yjm 512 -ytm 1024 batch/wordcount.jar