FlinkJob GeoLite2 Guide

Ordiy Lv5
  2024-12-10 17:51:37 2024-12-10 17:51:37 Created   2026-03-04 13:43:27 2026-03-04 13:43:27 Updated      

GeoLite2

GeoLite 离线文件下载地址:

1
2
3
4
5
https://github.com/P3TERX/GeoLite.mmdb

# GeoLite2-ASN.mmdb
# GeoLite2-City.mmdb
# GeoLite2-Country.mmdb

Country Reponse data 说明

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
{
   "continent":{
      "code":"NA",
      "geoname_id":6255149,
      "names":{
      }
   },
   "country":{
      "geoname_id":6252001,
      "is_in_european_union":false,
      "iso_code":"US",
      "names":{
      }
   },
   "maxmind":{
      
   },
   "registered_country":{
      "geoname_id":6252001,
      "is_in_european_union":false,
      "iso_code":"US",
      "names":{
      }
   },
   "represented_country":{
      "is_in_european_union":false
   },
   "traits":{
      "ip_address":"128.101.101.101",
      "is_anonymous":false,
      "is_anonymous_proxy":false,
      "is_anonymous_vpn":false,
      "is_anycast":false,
      "is_hosting_provider":false,
      "is_legitimate_proxy":false,
      "is_public_proxy":false,
      "is_residential_proxy":false,
      "is_satellite_provider":false,
      "is_tor_exit_node":false,
      "network":"128.101.0.0/16"
   }
}

Geo City Reponse

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
{
  "continent": {
    "code": "NA",
    "geoname_id": 123456,
    "names": {
    }
  },
  "country": {
    "geoname_id": 6252001,
    "is_in_european_union": false,
    "iso_code": "US",
    "names": {
    }
  },
  "maxmind": {
    "queries_remaining": 54321
  },
  "registered_country": {
    "geoname_id": 6252001,
    "is_in_european_union": true,
    "iso_code": "US",
    "names": {
    }
  },
  "represented_country": {
    "geoname_id": 6252001,
    "is_in_european_union": true,
    "iso_code": "US",
    "names": {
    },
    "type": "military"
  },
  "traits": {
    "ip_address": "1.2.3.4",
    "is_anycast": true,
    "network": "1.2.3.0/24",
    "autonomous_system_number": 1239,
    "autonomous_system_organization": "Linkem IR WiMax Network",
    "connection_type": "Cable/DSL",
    "domain": "example.com",
    "isp": "Linkem spa",
    "mobile_country_code": "310",
    "mobile_network_code": "004",
    "organization": "Linkem IR WiMax Network"
  },
  "city": {
    "geoname_id": 54321,
    "names": {
      "en": "Los Angeles"
    }
  },
  "location": {
    "accuracy_radius": 20,
    "latitude": 37.6293,
    "longitude": -122.1163,
    "metro_code": 807,
    "time_zone": "America/Los_Angeles"
  },
  "postal": {
    "code": "90001"
  },
  "subdivisions": [
    {
      "geoname_id": 5332921,
      "iso_code": "CA",
      "names": {
        "en": "California"
      }
    }
  ]
}

GeoLite2 ISP 字段信息

1
2
3
4

network,isp,organization,autonomous_system_number,autonomous_system_organization,mobile_country_code,mobile_network_code
2001:1700::/27,,,6730,"Sunrise Communications AG",,

GetLite2 数据读取

java read demo

https://github.com/maxmind/GeoIP2-java

IPGeoinfo API servcie

个人使用spring-boot实现的ip-geo-info servcie, 源代码:

https://github.com/ordiy/my-public-repository-collect/tree/main/java-api-service-ip-to-geo-info

特点：

  • 定义异步原子刷新GeoLite2 databaseSet(高QPS场景下 不会因为IP库更新出现 error)
  • 引入了guava load cache 提高缓存加载效率，同时使用cache提高服务QPS
  • 高性能高响应速度：4C8G instance 可实现 20K+ QPS
  • repsonse 简单，可以快速用于FlinkSQL UDF

在Flink 中使用

1
2
3
4
5
// tabEnv 定义UDF  URL 
      String urlConfigIpGeoInfo = parameterTool.get("url_config.ip_geo_info_server", "http://127.0.0.1:18901/data_v1/ip_geo_info/");
      MyIP2GeoInfoUDF myIP2GeoInfoUDF = new MyIP2GeoInfoUDF(urlConfigIpGeoInfo);
      tableEnv.createTemporaryFunction("ip_geo_info", myIP2GeoInfoUDF);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
        
// UDF IpGeoInfo        
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.functions.FunctionContext;
import org.apache.flink.table.functions.ScalarFunction;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.HashMap;
import java.util.Map;

import org.apache.flink.types.Row;

@FunctionHint(
        output = @DataTypeHint("ROW<c_lat DOUBLE, c_lan DOUBLE, c_country STRING, c_geokey STRING>")
)
public class MyIP2GeoInfoUDF extends ScalarFunction {
    private static final ObjectMapper mapper = new ObjectMapper();
    private final Map<String, Row> cache = new HashMap<>();
    private  String host_url ;


    public MyIP2GeoInfoUDF(String hostUrl) {
        this.host_url = hostUrl;
    }

    @Override
    public void open(FunctionContext context) throws Exception {
        super.open(context);
    }

    public Row eval(String ip) {
        if (ip == null || ip.equals("-")) {
            return Row.of(0.0, 0.0, "", "");
        }

        if (cache.containsKey(ip)) {
            return cache.get(ip);
        }

        HttpURLConnection conn = null;
        try {
            String urlString = host_url + ip;
            URL url = new URL(urlString);
            conn = (HttpURLConnection) url.openConnection();
            conn.setRequestMethod("GET");
            conn.setConnectTimeout(5000);
            conn.setReadTimeout(5000);

            int responseCode = conn.getResponseCode();
            if (responseCode == HttpURLConnection.HTTP_OK) {
                BufferedReader in = new BufferedReader(
                        new InputStreamReader(conn.getInputStream()));
                StringBuilder response = new StringBuilder();
                String inputLine;
                while ((inputLine = in.readLine()) != null) {
                    response.append(inputLine);
                }
                in.close();

                JsonNode jsonNode = mapper.readTree(response.toString());
                Row result = Row.of(
                        jsonNode.get("lat").asDouble(),
                        jsonNode.get("lon").asDouble(),
                        jsonNode.get("country_code").asText(),
                        jsonNode.get("geo_id").asText()
                );
                cache.put(ip, result);
                return result;
            } else {
                Row error = Row.of(0.0, 0.0, "Error: HTTP " + responseCode, "");
                cache.put(ip, error);
                return error;
            }
        } catch (Exception e) {
            Row error = Row.of(0.0, 0.0, "Error: " + e.getMessage(), "");
            cache.put(ip, error);
            return error;
        } finally {
            if (conn != null) {
                conn.disconnect();
            }
        }
    }
}

将GeoLite2

参考

https://dev.maxmind.com/geoip/docs/web-services/responses

  • Title: FlinkJob GeoLite2 Guide
  • Author: Ordiy
  • Created at : 2024-12-10 17:51:37
  • Updated at : 2026-03-04 13:43:27
  • Link: https://ordiy.github.io/posts/2024-12-10-geolite2-lib-action-guide-md/
  • License: This work is licensed under CC BY-NC-SA 4.0.
Comments
On this page
FlinkJob GeoLite2 Guide
  1. GeoLite2
    1. Country Reponse data 说明
    2. Geo City Reponse
    3. GeoLite2 ISP 字段信息
  2. GetLite2 数据读取
    1. java read demo
    2. IPGeoinfo API servcie
  3. 在Flink 中使用
    1. Flink UDF 使用
    2. 将GeoLite2
  4. 参考