View Categories

How to Enrich CSV Data with IP2Location Geolocation Using Apache Beam in Python

hexasoft
Updated on June 25, 2026

4 min read

How to Enrich CSV Data with IP2Location Geolocation Using Apache Beam in Python

Intro #

Apache Beam is an open-source data processing framework for building batch and streaming data pipelines. Instead of writing a simple script that processes records one by one, Beam lets you define a pipeline made up of multiple processing steps, such as reading input, parsing records, transforming data, enriching records, and writing the final output.

In this tutorial, we will use Apache Beam to read a CSV file, extract the IP address from each row, query the IP2Location Python library, append geolocation fields such as country, region, city, latitude, longitude, time zone, and ISP, then write the enriched records back into CSV format.

The pipeline flow will look like this:

Steps to Enrich CSV Data with IP2Location

For local testing, we will use Apache Beam’s DirectRunner, which runs the pipeline directly on the same machine. This is suitable for tutorials, development, testing, and small to medium local batch jobs. Later, the same Beam pipeline concept can be adapted to run on distributed runners such as Google Cloud Dataflow, Apache Flink, or Apache Spark.

Prerequisites #

This tutorial will be using a Debian 13 server, hence some of the steps you’ll see will be specific to that operating system.

The below items are also required for our example:

You can use either the free IP2Location LITE BIN database or a commercial IP2Location BIN database, depending on the fields you need.

Step 1: Update Debian packages #

Run the following commands on your server:

💻 
sudo apt update
sudo apt upgrade -y

Step 2: Install Python and build tools #

Install Python, virtual environment support, development headers, and basic build tools:

💻 
sudo apt install -y python3 python3-venv python3-dev python3-pip build-essential

Check the Python version:

💻 
python3 --version

Step 3: Create a project directory #

Create a directory for this tutorial:

💻 
mkdir ip2location-beam-csv
cd ip2location-beam-csv

Create folders for output, and database files:

💻 
mkdir data output

Step 4: Create and activate a Python virtual environment #

Create a virtual environment:

💻 
python3 -m venv .venv

Activate it:

💻 
source .venv/bin/activate

Upgrade pip:

💻 
pip install --upgrade pip

Step 5: Install Apache Beam and IP2Location Python #

Install the required Python packages:

💻 
pip install apache-beam IP2Location

Step 6: Prepare the IP2Location BIN database #

If you don’t have an account, you can either sign up for the free LITE BIN database or subscribe to the commercial BIN database. Upon login to the user dashboard, navigate to the Download section and download the zipped file containing the BIN file that you want. Then, extract the IP2Location BIN file and place it inside the data directory.

For our case, we’ll download the IP2Location DB26 IPv6 BIN database and place it inside the data directory.

E.g., data/IPV6-COUNTRY-REGION-CITY-LATITUDE-LONGITUDE-ZIPCODE-TIMEZONE-ISP-DOMAIN-NETSPEED-AREACODE-WEATHER-MOBILE-ELEVATION-USAGETYPE-ADDRESSTYPE-CATEGORY-DISTRICT-ASN.BIN

The exact BIN database edition determines which geolocation fields are available. For example, a country-only BIN database may only return country information, while a more complete database can return region, city, latitude, longitude, ZIP code, time zone, ISP, and other fields.

Step 7: Prepare a sample CSV file #

Create a file named orders.csv:

💻 
nano orders.csv

Add the following sample data:

orders.csv 
order_id,ip,amount
1001,8.8.8.8,59.90
1002,1.1.1.1,120.00
1003,208.67.222.222,35.50
1004,,10.00

Save and close the file.

In this example, the ip column contains the IP address that we want to enrich with geolocation data.

Step 8: Create the Apache Beam enrichment script #

Create a Python file:

💻 
nano enrich_csv_ip2location_beam.py

Add the following code:

enrich_csv_ip2location_beam.py 
import argparse
import csv
import io

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

import IP2Location


INPUT_COLUMNS = [
    "order_id",
    "ip",
    "amount",
]

OUTPUT_COLUMNS = [
    "order_id",
    "ip",
    "amount",
    "country_code",
    "country_name",
    "region_name",
    "city_name",
    "latitude",
    "longitude",
    "zip_code",
    "time_zone",
    "isp",
    "domain",
    "net_speed",
    "idd_code",
    "area_code",
    "weather_station_code",
    "weather_station_name",
    "mcc",
    "mnc",
    "mobile_brand",
    "elevation",
    "usage_type",
    "address_type",
    "category",
    "district",
    "asn",
    "as",
    "as_domain",
    "as_usage_type",
    "as_cidr",
    "geo_error",
]

def parse_csv_line(line):
    """
    Convert one CSV line into a Python dictionary.

    Example:
    1001,8.8.8.8,59.90

    Becomes:
    {
        "order_id": "1001",
        "ip": "8.8.8.8",
        "amount": "59.90"
    }
    """
    values = next(csv.reader([line]))
    return dict(zip(INPUT_COLUMNS, values))


class IP2LocationEnrichDoFn(beam.DoFn):
    """
    Apache Beam DoFn for enriching each record with IP2Location geolocation data.
    """

    def __init__(self, db_path, ip_field="ip"):
        self.db_path = db_path
        self.ip_field = ip_field
        self.database = None

    def setup(self):
        """
        Load the IP2Location BIN database.

        Beam calls setup() when the worker initializes this DoFn instance.
        This avoids opening the BIN database repeatedly for every row.
        """
        self.database = IP2Location.IP2Location(self.db_path)

    def process(self, record):
        """
        Enrich one input record and emit one output record.
        """
        ip = (record.get(self.ip_field) or "").strip()

        record["country_code"] = ""
        record["country_name"] = ""
        record["region_name"] = ""
        record["city_name"] = ""
        record["latitude"] = ""
        record["longitude"] = ""
        record["zip_code"] = ""
        record["time_zone"] = ""
        record["isp"] = ""
        record["domain"] = ""
        record["net_speed"] = ""
        record["idd_code"] = ""
        record["area_code"] = ""
        record["weather_station_code"] = ""
        record["weather_station_name"] = ""
        record["mcc"] = ""
        record["mnc"] = ""
        record["mobile_brand"] = ""
        record["elevation"] = ""
        record["usage_type"] = ""
        record["address_type"] = ""
        record["category"] = ""
        record["district"] = ""
        record["asn"] = ""
        record["as"] = ""
        record["as_domain"] = ""
        record["as_usage_type"] = ""
        record["as_cidr"] = ""
        record["geo_error"] = ""

        if not ip:
            record["geo_error"] = "missing_ip"
            yield record
            return

        try:
            geo = self.database.get_all(ip)

            record["country_code"] = geo.country_short
            record["country_name"] = geo.country_long
            record["region_name"] = geo.region
            record["city_name"] = geo.city
            record["latitude"] = geo.latitude
            record["longitude"] = geo.longitude
            record["zip_code"] = geo.zipcode
            record["time_zone"] = geo.timezone
            record["isp"] = geo.isp
            record["domain"] = geo.domain
            record["net_speed"] = geo.netspeed
            record["idd_code"] = geo.idd_code
            record["area_code"] = geo.area_code
            record["weather_station_code"] = geo.weather_code
            record["weather_station_name"] = geo.weather_name
            record["mcc"] = geo.mcc
            record["mnc"] = geo.mnc
            record["mobile_brand"] = geo.mobile_brand
            record["elevation"] = geo.elevation
            record["usage_type"] = geo.usage_type
            record["address_type"] = geo.address_type
            record["category"] = geo.category
            record["district"] = geo.district
            record["asn"] = geo.asn
            record["as"] = geo.as_name
            record["as_domain"] = geo.as_domain
            record["as_usage_type"] = geo.as_usagetype
            record["as_cidr"] = geo.as_cidr

        except Exception as error:
            record["geo_error"] = str(error)

        yield record


def record_to_csv_line(record):
    """
    Convert a Python dictionary back into one CSV line.

    csv.DictWriter is used instead of string concatenation so that commas,
    quotes, and special characters are escaped properly.
    """
    output = io.StringIO()
    writer = csv.DictWriter(
        output,
        fieldnames=OUTPUT_COLUMNS,
        extrasaction="ignore",
    )
    writer.writerow(record)

    return output.getvalue().strip("\r\n")


def run(argv=None):
    parser = argparse.ArgumentParser()

    parser.add_argument(
        "--input",
        required=True,
        help="Path to the input CSV file.",
    )

    parser.add_argument(
        "--output",
        required=True,
        help="Output file prefix. Beam will create the final CSV file using this prefix.",
    )

    parser.add_argument(
        "--db",
        required=True,
        help="Path to the IP2Location BIN database.",
    )

    known_args, pipeline_args = parser.parse_known_args(argv)

    pipeline_options = PipelineOptions(pipeline_args)

    with beam.Pipeline(options=pipeline_options) as pipeline:
        (
            pipeline
            | "Read CSV" >> beam.io.ReadFromText(
                known_args.input,
                skip_header_lines=1,
            )
            | "Parse CSV to Dict" >> beam.Map(parse_csv_line)
            | "Enrich with IP2Location" >> beam.ParDo(
                IP2LocationEnrichDoFn(
                    db_path=known_args.db,
                    ip_field="ip",
                )
            )
            | "Convert Dict to CSV Line" >> beam.Map(record_to_csv_line)
            | "Write Enriched CSV" >> beam.io.WriteToText(
                known_args.output,
                file_name_suffix=".csv",
                num_shards=1,
                header=",".join(OUTPUT_COLUMNS),
            )
        )


if __name__ == "__main__":
    run()

Save and close the file.

Step 9: Run the Apache Beam pipeline locally #

Run the pipeline with DirectRunner:

💻 
python enrich_csv_ip2location_beam.py \
  --input orders.csv \
  --output output/orders_enriched \
  --db data/IPV6-COUNTRY-REGION-CITY-LATITUDE-LONGITUDE-ZIPCODE-TIMEZONE-ISP-DOMAIN-NETSPEED-AREACODE-WEATHER-MOBILE-ELEVATION-USAGETYPE-ADDRESSTYPE-CATEGORY-DISTRICT-ASN.BIN \
  --runner=DirectRunner

Beam will read the input CSV file, enrich each row with geolocation information, and write the result to the output directory.

Because we used num_shards=1, the output file should look similar to this:

output/orders_enriched-00000-of-00001.csv

Step 10: View the enriched CSV output #

Display the output file:

💻 
cat output/orders_enriched-00000-of-00001.csv

Example output:

orders_enriched-00000-of-00001.csv 
order_id,ip,amount,country_code,country_name,region_name,city_name,latitude,longitude,zip_code,time_zone,isp,domain,net_speed,idd_code,area_code,weather_station_code,weather_station_name,mcc,mnc,mobile_brand,elevation,usage_type,address_type,category,district,asn,as,as_domain,as_usage_type,as_cidr,geo_error
1001,8.8.8.8,59.90,US,United States of America,California,Mountain View,37.386051,-122.083847,94035,-07:00,Google LLC,google.com,T1,1,650,USCA0746,Mountain View,-,-,-,32,DCH,A,IAB19-11,Santa Clara County,15169,Google LLC,google.com,DCH,8.8.8.0/24,
1002,1.1.1.1,120.00,AU,Australia,Queensland,Brisbane,-27.467541,153.028091,4000,+10:00,APNIC and CloudFlare DNS Resolver Project,cloudflare.com,T1,61,07,ASXX0016,Brisbane,-,-,-,16,CDN,A,IAB19-11,Brisbane,13335,CloudFlare Inc,cloudflare.com,CDN,1.1.1.0/24,
1003,208.67.222.222,35.50,US,United States of America,California,San Jose,37.339390,-121.894958,95134,-07:00,Cisco OpenDNS LLC,opendns.com,T1,1,408,USCA0993,San Jose,-,-,-,24,CDN,A,IAB19-11,Santa Clara County,36692,Cisco OpenDNS LLC,opendns.com,CDN,208.67.222.0/24,
1004,,10.00,,,,,,,,,,,,,,,,,,,,,,,,,,,,,missing_ip

Since our example is relying on the DB26 IPv6 BIN file, you will see the columns contained in that BIN file.

How the pipeline works #

The pipeline is made up of several Beam transforms:

enrich_csv_ip2location_beam.py 
pipeline
| "Read CSV" >> beam.io.ReadFromText(...)
| "Parse CSV to Dict" >> beam.Map(parse_csv_line)
| "Enrich with IP2Location" >> beam.ParDo(IP2LocationEnrichDoFn(...))
| "Convert Dict to CSV Line" >> beam.Map(record_to_csv_line)
| "Write Enriched CSV" >> beam.io.WriteToText(...)

Read CSV #

enrich_csv_ip2location_beam.py 
beam.io.ReadFromText(...)

This reads the CSV file line by line. The header row is skipped because we already define the expected input columns in the script.

Parse CSV to dictionary #

enrich_csv_ip2location_beam.py 
beam.Map(parse_csv_line)

This converts each CSV row into a Python dictionary.

For example:

orders.csv 
1001,8.8.8.8,59.90

becomes:

{
    "order_id": "1001",
    "ip": "8.8.8.8",
    "amount": "59.90"
}

Using a dictionary makes the enrichment code easier to read because we can access the IP address by name:

enrich_csv_ip2location_beam.py 
ip = record.get("ip")

Enrich with IP2Location #

enrich_csv_ip2location_beam.py 
beam.ParDo(IP2LocationEnrichDoFn(...))

This step uses the IP2Location Python library to look up geolocation information for each IP address.

The IP2Location BIN database is loaded inside the setup() method:

enrich_csv_ip2location_beam.py 
def setup(self):
    self.database = IP2Location.IP2Location(self.db_path)

This is better than opening the BIN file inside process() because process() runs for every row. Loading the database in setup() allows each worker instance to reuse the same database object.

Write back to CSV #

After the record has been enriched, the script converts the dictionary back into a CSV line:

enrich_csv_ip2location_beam.py 
beam.Map(record_to_csv_line)

Finally, Beam writes the enriched data into a CSV file:

enrich_csv_ip2location_beam.py 
beam.io.WriteToText(...)

Handling missing IP addresses #

In the sample code, rows with missing IP addresses are not removed. Instead, they are kept in the output file with the geo_error field set to missing_ip.

For example:

orders_enriched-00000-of-00001.csv 
1004,,10.00,,,,,,,,,,,,,,,,,,,,,,,,,,,,,missing_ip

If you prefer to skip rows without IP addresses, you can change this part:

enrich_csv_ip2location_beam.py 
if not ip:
    record["geo_error"] = "missing_ip"
    yield record
    return

to:

enrich_csv_ip2location_beam.py 
if not ip:
    return

When a Beam process() function returns without yield, that record is not emitted to the next step and will not appear in the final output.

Notes for production use #

For a tutorial, DirectRunner and num_shards=1 make the output easy to understand.

For production workloads, consider the following:

  • Use multiple output shards for better parallel write performance.
  • Store input and output files in a durable storage system.
  • Monitor failed or missing IP lookups.
  • Use a recent IP2Location BIN database.
  • Choose a BIN database edition that contains the fields you need.
  • Avoid calling a remote API for every row if you process large files.

Using the local IP2Location BIN database is usually more efficient for batch enrichment because the lookup happens locally without making an external API request for every row.

Conclusion #

Apache Beam provides a clean way to build data processing pipelines in Python. In this tutorial, we used Beam to read CSV data, parse each row into a dictionary, enrich the row with IP2Location geolocation data, and write the enriched result back into CSV format.

This approach keeps the pipeline easy to understand:

Read → Parse → Enrich → Format → Write

Although this example runs locally on Debian 13 with DirectRunner, the same Beam pipeline structure can be adapted later for larger data processing environments.

Updated on June 25, 2026
Apache IP geolocation
Scroll to Top