Flink入门wordCount
Flink的编程模型
1、获取Flink上下文;
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
2、加载、创建数据;
DataSet
3、数据转换;
Transformation
4、数据结果存放;
5、触发执行。
env.execution
创新互联公司专注于企业营销型网站建设、网站重做改版、鄂尔多斯网站定制设计、自适应品牌网站建设、H5网站设计、购物商城网站建设、集团公司官网建设、外贸营销网站建设、高端网站制作、响应式网页设计等建站业务,价格优惠性价比高,为鄂尔多斯等各大城市提供网站开发制作服务。
下面为flink输出wordcount数据:
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;
public class FlinkMain {
@SuppressWarnings("serial")
public static class LineSplit implements FlatMapFunction>{
@SuppressWarnings("rawtypes")
@Override
/**
* @param value 原数据
* @param out 输出的数据
*/
public void flatMap(String value, Collector> out) throws Exception {
String[] tokens = value.split(" ");
for (String token : tokens) {
if(token!=null && token.length()>0){
Tuple2 t = new Tuple2(token,1);
out.collect(t);
}
}
}
}
public static void main(String[] args) throws Exception {
//创建flink上下文
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
//创建数据集
DataSet text = env.fromElements("to be","or no to be","is question");
//对数据集转换
DataSet> count = text.flatMap(new LineSplit());
//输出转换后的数据集(print中包含了env.execute执行)
count.print();
System.out.println("-----------------------");
//对数据集分组统计转换,0,1是下标,对应Tuple2类中的参数
count = count.groupBy(0).sum(1);
//控制台输出数据集
count.print();
System.out.println("-----------------------");
}
}
Flink使用sql方式转换数据
import java.util.ArrayList;
import java.util.List;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.java.BatchTableEnvironment;
public class FlinkMain2 {
@SuppressWarnings({ "unchecked", "rawtypes" })
public static void main(String[] args) throws Exception {
//创建flink上下文
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tEnv = TableEnvironment.getTableEnvironment(env);
List list = new ArrayList();
String workStr = "to be or no to be is question";
String[] tokens = workStr.split(" ");
for (String token : tokens) {
if(token!=null && token.length()>0){
list.add( new WordCount(token,1));
}
}
//创建数据集
DataSet input = env.fromCollection(list);
//注册为数据表wordCount为数据库表,word,frequency为wordCount表字段
tEnv.registerDataSet("wordCount", input, "word, frequency");
Table table = tEnv.sqlQuery(" SELECT word, SUM(frequency) as frequency FROM wordCount GROUP BY word" );
DataSet res = tEnv.toDataSet(table, WordCount.class);
//控制台输出
res.print();
}
public static class WordCount {
public String word;
public long frequency;
public WordCount(){}
public WordCount(String word, long frequency) {
this.word = word;
this.frequency = frequency;
}
@Override
public String toString() {
return "词语:" + word + ",词频:" + frequency;
}
}
}
新闻名称:Flink入门wordCount
当前URL:http://scyanting.com/article/jiicop.html