Real-Time IP Geolocation Enrichment with Apache Flink and IP2Location Java

cat > src/main/java/com/example/IP2LocationEnrichmentJob.java <<'EOF'
package com.example;

import com.ip2location.IP2Location;
import com.ip2location.IPResult;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

public class IP2LocationEnrichmentJob {

    public static void main(String[] args) throws Exception {
        if (args.length < 3) {
            System.err.println("Usage: IP2LocationEnrichmentJob <inputDir> <outputDir> <binFile>");
            System.err.println("Example:");
            System.err.println("  flink run target/ip2location-flink-demo.jar " +
                    "/home/user/flink-ip2location-demo/input " +
                    "/home/user/flink-ip2location-demo/output " +
                    "/opt/ip2location/IPV6-COUNTRY-REGION-CITY-LATITUDE-LONGITUDE-ZIPCODE-TIMEZONE-ISP-DOMAIN-NETSPEED-AREACODE-WEATHER-MOBILE-ELEVATION-USAGETYPE-ADDRESSTYPE-CATEGORY-DISTRICT-ASN.BIN");
            return;
        }

        String inputDir = args[0];
        String outputDir = args[1];
        String binFile = args[2];

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // Keep the tutorial simple and output easier to read.
        env.setParallelism(1);

        // Required for FileSink to finalize output files in streaming mode.
        env.enableCheckpointing(5000);

        FileSource<String> source = FileSource
                .forRecordStreamFormat(new TextLineInputFormat(), new Path(inputDir))
                .monitorContinuously(Duration.ofSeconds(2))
                .build();

        DataStream<String> lines = env.fromSource(
                source,
                WatermarkStrategy.noWatermarks(),
                "csv-file-source"
        );

        DataStream<String> enriched = lines
                .filter(line -> line != null && !line.trim().isEmpty())
                .filter(line -> !line.startsWith("event_id,"))
                .map(new EnrichWithIP2Location(binFile));

        // Print to TaskManager log for easy demo.
        enriched.print();

        FileSink<String> sink = FileSink
                .forRowFormat(new Path(outputDir), new SimpleStringEncoder<String>("UTF-8"))
                .build();

        enriched.sinkTo(sink);

        env.execute("IP2Location Geolocation Enrichment Demo");
    }

    public static class EnrichWithIP2Location extends RichMapFunction<String, String> {
        private final String binFile;
        private transient IP2Location ip2Location;

        public EnrichWithIP2Location(String binFile) {
            this.binFile = binFile;
        }

        @Override
        public void open(OpenContext openContext) throws Exception {
            ip2Location = new IP2Location();
            ip2Location.Open(binFile, true);
        }

        @Override
        public String map(String line) throws Exception {
            String[] fields = line.split(",", -1);

            if (fields.length < 5) {
                return escape(line) + ",INVALID_RECORD,,,,,,,";
            }

            String eventId = fields[0].trim();
            String eventTime = fields[1].trim();
            String ipAddress = fields[2].trim();
            String pageUrl = fields[3].trim();
            String userAgent = fields[4].trim();

            IPResult result = ip2Location.IPQuery(ipAddress);

            String status = result.getStatus();

            if (!"OK".equals(status)) {
                return String.join(",",
                        escape(eventId),
                        escape(eventTime),
                        escape(ipAddress),
                        escape(pageUrl),
                        escape(userAgent),
                        escape(status),
                        "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", ""
                );
            }

            return String.join(",",
                    escape(eventId),
                    escape(eventTime),
                    escape(ipAddress),
                    escape(pageUrl),
                    escape(userAgent),
                    escape(status),
                    escape(result.getCountryShort()),
                    escape(result.getCountryLong()),
                    escape(result.getRegion()),
                    escape(result.getCity()),
                    escape(result.getISP()),
                    String.valueOf(result.getLatitude()),
                    String.valueOf(result.getLongitude()),
                    escape(result.getDomain()),
                    escape(result.getZipCode()),
                    escape(result.getTimeZone()),
                    escape(result.getNetSpeed()),
                    escape(result.getIDDCode()),
                    escape(result.getAreaCode()),
                    escape(result.getWeatherStationCode()),
                    escape(result.getWeatherStationName()),
                    escape(result.getMCC()),
                    escape(result.getMNC()),
                    escape(result.getMobileBrand()),
                    String.valueOf(result.getElevation()),
                    escape(result.getUsageType()),
                    escape(result.getAddressType()),
                    escape(result.getCategory()),
                    escape(result.getDistrict()),
                    escape(result.getASN()),
                    escape(result.getAS()),
                    escape(result.getASDomain()),
                    escape(result.getASUsageType()),
                    escape(result.getASCIDR())
            );
        }

        @Override
        public void close() {
            if (ip2Location != null) {
                ip2Location.Close();
            }
        }

        private static String escape(String value) {
            if (value == null) {
                return "";
            }

            String escaped = value.replace("\"", "\"\"");

            if (escaped.contains(",") || escaped.contains("\"") || escaped.contains("\n")) {
                return "\"" + escaped + "\"";
            }

            return escaped;
        }
    }
}
EOF