一、需求分析
1.kafkaSource根据经纬度查询高德API关联位置信息
2.查询一条数据及时没有及时的返回，也可以异步Flink IO
1.1 数据样式
user001,A1,2020-10-10 10:10:10,1,115.963,39.3659 user002,A2,2020-10-10 10:10:10,1,123.544,36.2356 user003,A3,2020-10-10 10:10:10,1,123.568,47.3215
二、高德地图key准备
2.1 进入开发者中心
2.2 引入http的依赖
<!--引入httpClient的依赖-->
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpclient</artifactId>
<version>4.5.6</version>
</dependency>
三、Flink开发
3.1 ActivityBean.Java
public class ActivityBean {
public String uid;
public String aid;
public String activityName;
public String time;
public int eventType;
public String province;
public double lonGitude;
public double latitude;
public ActivityBean() {
}
public ActivityBean(String uid, String aid, String activityName, String time, int eventType, double longitude,double latitude,String province) {
this.uid = uid;
this.aid = aid;
this.activityName = activityName;
this.time = time;
this.eventType = eventType;
this.province = province;
this.latitude = latitude;
this.longitude = longitude;
}
@Override
public String toString() {
return "ActivityBean{" +
"uid='" + uid + ''' +
", aid='" + aid + ''' +
", activityName='" + activityName + ''' +
", time='" + time + ''' +
", eventType=" + eventType +
", province='" + province + ''' +
", longitude=" + longitude +
", latitude=" + latitude +
'}';
}
public static ActivityBean of(String uid,String aid,String activityName,String time,int eventType,double longitude,double latitude,String province){
return new ActivityBean(uid,aid,activityName,time,eventType,longitude,latitude,province);
}
}
3.2 RichMapFunction.java
public class GeoToActivityBeanFunction extends RichMapFunction<String,ActivityBean> {
private CloseableHttpClient httpClient = null;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
httpClient = HttpClients.createDefault();
}
@Override
public ActivityBean map(String line) throws Exception {
String[] fields = line.split(",");
String uid = fields[0];
String aid = fields[1];
String time = fields[2];
int eventType = Integer.parseInt(fields[3]);
String longitude = fields[4];
String latitude = fields[5];
String url = "https://restapi.amap.com/v3/geocode/regeo?key=32451f2e804ded152cef65c3cd16452d&location=" + longitude + "," + latitude;
String province = null;
HttpGet httpGet = new HttpGet(url);
CloseableHttpResponse response = httpClient.execute(httpGet);
try{
int status = response.getStatusLine().getStatusCode();
if (status == 200) {
//获取请求的json字符串
String result = EntityUtils.toString(response.getEntity());
System.out.println(result);
JSONObject jsonObject = JSONObject.parseObject(result);
JSONObject regeocode = jsonObject.getJSONObject("regeocode");
if (regeocode != null && !regeocode.isEmpty()) {
JSONObject address = regeocode.getJSONObject("addressComponent");
//获取省市区
province = address.getString("province");
//String city = address.getString("city");
//String businessAreas = address.getString("businessAreas");
}
}
}finally {
response.close();
}
return ActivityBean.of(uid,aid,"未查询数据库",time,eventType,Double.parseDouble(longitude),Double.parseDouble(latitude),province);
}
@Override
public void close() throws Exception {
super.close();
httpClient.close();
}
}
3.3 Flink执行程序
public class GeoQueryLocationActivityCount {
public static void main(String[] args) throws Exception {
//1.获取环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//2.kafka配置
String topic = "activity";
Properties prop = new Properties();
prop.setProperty("bootstrap.servers", "192.168.52.200:9092");//多个的话可以指定
prop.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
prop.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
prop.setProperty("auto.offset.reset", "earliest");
prop.setProperty("group.id", "consumer1");
FlinkKafkaConsumer<String> myConsumer = new FlinkKafkaConsumer<String>(topic, new SimpleStringSchema(), prop);
//3.获取数据
DataStream<String> lines = env.addSource(myConsumer);
SingleOutputStreamOperator<ActivityBean> beans = lines.map(new GeoToActivityBeanFunction());
beans.print();
//7.执行
env.execute("GeoQueryLocationActivityCount");
}
}
3.4 测试
1.开启kafka生产者
bin/kafka-console-producer.sh --broker-list 192.168.52.200:9092,1.168.52.201:9092,192.168.52.202:9092 --topic activity
测试结果：
这里我们想一个问题，每次flink查询计算时候，需要依赖于查询高德地图的返回结果，如果没有返回结果，那么是不是flink程序就会一直卡在这里，不会执行下去
四、Flink异步Async I/O
4.1 引入异步的httpclient依赖
<!--引入高效的异步httpClient的依赖-->
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpasyncclient</artifactId>
<version>4.1.4</version>
</dependency>
4.2 RichAsyncFunction.java
public class AsyncGeoToActivityBeanFunction extends RichAsyncFunction<String,ActivityBean> {
private transient CloseableHttpAsyncClient httpAsyncClient;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
//1.初始化异步的httpclient
RequestConfig requestConfig = RequestConfig.custom()
.setSocketTimeout(3000)
.setConnectTimeout(3000)
.build();
httpAsyncClient = HttpAsyncClients.custom()
.setMaxConnTotal(20)
.setDefaultRequestConfig(requestConfig).build();
httpAsyncClient.start();
}
@Override
public void asyncInvoke(String line, ResultFuture<ActivityBean> resultFuture) throws Exception {
String[] fields = line.split(",");
String uid = fields[0];
String aid = fields[1];
String time = fields[2];
int eventType = Integer.parseInt(fields[3]);
String longitude = fields[4];
String latitude = fields[5];
String url = "https://restapi.amap.com/v3/geocode/regeo?key=32451f2e804ded152cef65c3cd16452d&location=" + longitude + "," + latitude;
HttpGet httpGet = new HttpGet(url);
Future<HttpResponse> future = httpAsyncClient.execute(httpGet, null);
CompletableFuture.supplyAsync(new Supplier<String>() {
@Override
public String get() {
try{
HttpResponse response = future.get();
String province = null;
if (response.getStatusLine().getStatusCode() == 200) {
//获取请求的json字符串
String result = EntityUtils.toString(response.getEntity());
System.out.println(result);
JSONObject jsonObject = JSONObject.parseObject(result);
JSONObject regeocode = jsonObject.getJSONObject("regeocode");
if (regeocode != null && !regeocode.isEmpty()) {
JSONObject address = regeocode.getJSONObject("addressComponent");
//获取省市区
province = address.getString("province");
//String city = address.getString("city");
//String businessAreas = address.getString("businessAreas");
}
}
return province;
} catch (Exception e) {
return null;
}
}
}).thenAccept((String province) ->{
resultFuture.complete(Collections.singleton(ActivityBean.of(uid,aid,null,time,eventType,0,0,province)));
});
}
@Override
public void close() throws Exception {
super.close();
httpAsyncClient.close();
}
}
4.3 Flink主程序
public class AsyncGeoQueryLocationActivityCount {
public static void main(String[] args) throws Exception {
//1.获取环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//2.kafka配置
String topic = "activity";
Properties prop = new Properties();
prop.setProperty("bootstrap.servers", "192.168.52.200:9092");//多个的话可以指定
prop.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
prop.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
prop.setProperty("auto.offset.reset", "earliest");
prop.setProperty("group.id", "consumer1");
FlinkKafkaConsumer<String> myConsumer = new FlinkKafkaConsumer<String>(topic, new SimpleStringSchema(), prop);
//3.获取数据
DataStream<String> lines = env.addSource(myConsumer);
// 无序的异步
SingleOutputStreamOperator<ActivityBean> beans = AsyncDataStream.unorderedWait(lines, new AsyncGeoToActivityBeanFunction(), 0, TimeUnit.MICROSECONDS, 10);
beans.print();
//7.执行
env.execute("AsyncGeoQueryLocationActivityCount");
}
}
4.4 总结
异步请求MySQL 【线程池】
异步请求ES
异步请求…..
原文始发于微信公众号（Coding路人王）：【Flink-需求】KafkaSource根据经纬度查询高德API关联位置信息
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 举报，一经查实，本站将立刻删除。
文章由极客之音整理，本文链接：https://www.bmabk.com/index.php/post/41752.html
