fio-l1-nyc-csvwkt-addresses

Rerun

import pandas as pd
import geopandas as gpd
from shapely import wkt
from pyarrow import parquet as pq
import pyarrow as pa

# Read CSV
df = pd.read_csv("/work/nyc_addresses.csv")

# Parse WKT geometry — these are lon/lat points -> EPSG:4326
df["geometry"] = df["geometry_wkt"].apply(wkt.loads)
gdf = gpd.GeoDataFrame(df, geometry="geometry", crs="EPSG:4326")
gdf.drop(columns=["geometry_wkt"], inplace=True)

# Convert recorded_at to timestamp[us] (datetime64[us] in pandas / timestamp with microsecond resolution)
gdf["recorded_at"] = pd.to_datetime(gdf["recorded_at"])

# Convert unit_count to int32
gdf["unit_count"] = gdf["unit_count"].astype("int32")

# All other columns as string
string_cols = ["id", "country", "postcode", "street", "number", "unit", "postal_city"]
for col in string_cols:
    gdf[col] = gdf[col].astype(str)

# Build a PyArrow schema to enforce exact types
schema = pa.schema([
    ("id", pa.utf8()),
    ("country", pa.utf8()),
    ("postcode", pa.utf8()),
    ("street", pa.utf8()),
    ("number", pa.utf8()),
    ("unit", pa.utf8()),
    ("postal_city", pa.utf8()),
    ("recorded_at", pa.timestamp("us")),       # timestamp with microsecond precision
    ("unit_count", pa.int32()),
    ("geometry", pa.binary()),                   # GeoParquet stores geometry as WKB binary
])

# Convert to PyArrow Table and enforce schema
table = pa.Table.from_pandas(gdf, schema=schema, preserve_index=False)

# Write GeoParquet
pq.write_table(table, "/work/addresses.geoparquet")

print("✅ addresses.geoparquet written successfully")
print(f"   Row count: {len(gdf)}")
print(f"   Columns: {list(gdf.columns)}")
print(f"   dtypes:\n{gdf.dtypes}")

import pandas as pd
import geopandas as gpd
from shapely import wkt
import pyarrow as pa

# Read CSV
df = pd.read_csv("/work/nyc_addresses.csv")

# Parse WKT geometry — these are lon/lat points -> EPSG:4326
df["geometry"] = df["geometry_wkt"].apply(wkt.loads)
gdf = gpd.GeoDataFrame(df, geometry="geometry", crs="EPSG:4326")
gdf.drop(columns=["geometry_wkt"], inplace=True)

# Convert recorded_at to datetime
gdf["recorded_at"] = pd.to_datetime(gdf["recorded_at"])

# Convert unit_count to int32
gdf["unit_count"] = gdf["unit_count"].astype("int32")

# All other columns as string
string_cols = ["id", "country", "postcode", "street", "number", "unit", "postal_city"]
for col in string_cols:
    gdf[col] = gdf[col].astype(str)

# Build a PyArrow schema for non-geometry columns
# GeoPandas' to_parquet handles the geometry column internally.
schema = pa.schema([
    ("id", pa.utf8()),
    ("country", pa.utf8()),
    ("postcode", pa.utf8()),
    ("street", pa.utf8()),
    ("number", pa.utf8()),
    ("unit", pa.utf8()),
    ("postal_city", pa.utf8()),
    ("recorded_at", pa.timestamp("us")),
    ("unit_count", pa.int32()),
])

# Use GeoPandas to_parquet which properly handles geometry -> WKB in GeoParquet
gdf.to_parquet(
    "/work/addresses.geoparquet",
    schema=schema,
    index=False,
    geometry_encoding="WKB",
)

print("✅ addresses.geoparquet written successfully")
print(f"   Row count: {len(gdf)}")
print(f"   Columns: {list(gdf.columns)}")
print(f"   dtypes:\n{gdf.dtypes}")
print(f"   CRS: {gdf.crs}")

import pandas as pd
import geopandas as gpd
from shapely import wkt
import pyarrow as pa
import pyarrow.parquet as pq
import json

# Read CSV
df = pd.read_csv("/work/nyc_addresses.csv")

# Parse WKT geometry
df["geometry"] = df["geometry_wkt"].apply(wkt.loads)
gdf = gpd.GeoDataFrame(df, geometry="geometry", crs="EPSG:4326")
gdf.drop(columns=["geometry_wkt"], inplace=True)

# Convert recorded_at to datetime
gdf["recorded_at"] = pd.to_datetime(gdf["recorded_at"])

# Convert unit_count to int32
gdf["unit_count"] = gdf["unit_count"].astype("int32")

# All other columns as string
string_cols = ["id", "country", "postcode", "street", "number", "unit", "postal_city"]
for col in string_cols:
    gdf[col] = gdf[col].astype(str)

# Build table from pandas using GeoDataFrame's built-in WKB column approach
# First convert geometry to WKB bytes
import shapely
gdf["__geometry_wkb__"] = shapely.to_wkb(gdf.geometry.values)

# Drop geometry and rename wkb column
df_out = gdf.drop(columns=["geometry"]).copy()
df_out.rename(columns={"__geometry_wkb__": "geometry"}, inplace=True)

# Construct PyArrow schema
schema = pa.schema([
    ("id", pa.utf8()),
    ("country", pa.utf8()),
    ("postcode", pa.utf8()),
    ("street", pa.utf8()),
    ("number", pa.utf8()),
    ("unit", pa.utf8()),
    ("postal_city", pa.utf8()),
    ("recorded_at", pa.timestamp("us")),
    ("unit_count", pa.int32()),
    ("geometry", pa.binary()),
])

# Build the table
table = pa.Table.from_pydict(
    {col: df_out[col].values for col in df_out.columns},
    schema=schema,
)

# Write GeoParquet with geo metadata
pq.write_table(
    table,
    "/work/addresses.geoparquet",
    geo_metadata={
        "columns": {
            "geometry": {
                "encoding": "WKB",
                "geometry_type": "Point",
                "bbox": [float(gdf.total_bounds[0]), float(gdf.total_bounds[1]),
                         float(gdf.total_bounds[2]), float(gdf.total_bounds[3])],
                "crs": "EPSG:4326",
            }
        },
        "primary_column": "geometry",
        "version": "1.0.0",
    },
)

print("✅ addresses.geoparquet written successfully")
print(f"   Row count: {len(gdf)}")
print(f"   CRS: {gdf.crs}")

# Quick verification
df_check = gpd.read_parquet("/work/addresses.geoparquet")
print(f"   Verified: {len(df_check)} rows, CRS={df_check.crs}")
print(f"   dtypes:\n{df_check.dtypes}")

# Test the user's intended queries
import datetime
mask = df_check["recorded_at"] > datetime.datetime(2024, 1, 1, tzinfo=None)
print(f"   WHERE recorded_at > '2024-01-01': {mask.sum()} rows")
print(f"   SUM(unit_count): {df_check['unit_count'].sum()}")

import pandas as pd
import geopandas as gpd
from shapely import wkt, to_wkb
import pyarrow as pa
import pyarrow.parquet as pq
import json

# Read CSV
df = pd.read_csv("/work/nyc_addresses.csv")

# Parse WKT geometry
df["geometry"] = df["geometry_wkt"].apply(wkt.loads)
gdf = gpd.GeoDataFrame(df, geometry="geometry", crs="EPSG:4326")
gdf.drop(columns=["geometry_wkt"], inplace=True)

# Convert recorded_at to datetime
gdf["recorded_at"] = pd.to_datetime(gdf["recorded_at"])

# Convert unit_count to int32
gdf["unit_count"] = gdf["unit_count"].astype("int32")

# All other columns as string
string_cols = ["id", "country", "postcode", "street", "number", "unit", "postal_city"]
for col in string_cols:
    gdf[col] = gdf[col].astype(str)

# --- Build the table with exact types via PyArrow ---

# Geometry as WKB binary
wkb_col = to_wkb(gdf.geometry.values)

# Build column arrays with desired types
cols = {}
cols["id"] = pa.array(gdf["id"].values, type=pa.utf8())
cols["country"] = pa.array(gdf["country"].values, type=pa.utf8())
cols["postcode"] = pa.array(gdf["postcode"].values, type=pa.utf8())
cols["street"] = pa.array(gdf["street"].values, type=pa.utf8())
cols["number"] = pa.array(gdf["number"].values, type=pa.utf8())
cols["unit"] = pa.array(gdf["unit"].values, type=pa.utf8())
cols["postal_city"] = pa.array(gdf["postal_city"].values, type=pa.utf8())
cols["recorded_at"] = pa.array(gdf["recorded_at"].values, type=pa.timestamp("us"))
cols["unit_count"] = pa.array(gdf["unit_count"].values, type=pa.int32())
cols["geometry"] = pa.array(wkb_col, type=pa.binary())

s = pa.schema([
    ("id", pa.utf8()),
    ("country", pa.utf8()),
    ("postcode", pa.utf8()),
    ("street", pa.utf8()),
    ("number", pa.utf8()),
    ("unit", pa.utf8()),
    ("postal_city", pa.utf8()),
    ("recorded_at", pa.timestamp("us")),
    ("unit_count", pa.int32()),
    ("geometry", pa.binary()),
])

table = pa.table(cols, schema=s)

# Build GeoParquet metadata (version 1.0.0)
bbox = [float(gdf.total_bounds[0]), float(gdf.total_bounds[1]),
        float(gdf.total_bounds[2]), float(gdf.total_bounds[3])]

geo_meta = {
    "version": "1.0.0",
    "primary_column": "geometry",
    "columns": {
        "geometry": {
            "encoding": "WKB",
            "geometry_types": ["Point"],
            "bbox": bbox,
            "crs": {
                "type": "GeographicCRS",
                "properties": {
                    "name": "EPSG:4326"
                }
            }
        }
    }
}

# Write with geo metadata in the key-value metadata
pq.write_table(
    table,
    "/work/addresses.geoparquet",
    metadata_collector=None,
)

# Now read back and attach geo metadata
pf = pq.ParquetFile("/work/addresses.geoparquet")
existing_meta = pf.metadata.metadata or {}
existing_meta[b"geo"] = json.dumps(geo_meta).encode("utf-8")

# Rewrite with the metadata
# We need to write the table again with the metadata
pq.write_table(
    table,
    "/work/addresses_tmp.geoparquet",
    metadata_collector=None,
)

# Use parquet's write_metadata or simply overwrite with the metadata dict
from pyarrow import parquet as pq2
# Actually let's use the write_table with the metadata dict approach properly
import os
os.remove("/work/addresses_tmp.geoparquet")

# Write directly with metadata
pq.write_table(
    table,
    "/work/addresses.geoparquet",
    metadata_collector=None,
)

# Re-open and add geo metadata
# We can write a rewritten metadata using the FileMetaData level
# Let's use a simpler approach: write to buffer with metadata
import io

# Actually the cleanest way: write table, then use ParquetWriter with the metadata
# Let's just do it properly
with pq.ParquetWriter(
    "/work/addresses.geoparquet",
    schema=table.schema,
    version="2.6",
) as writer:
    writer.write_table(table)

# Now read metadata and add geo key
pf = pq.ParquetFile("/work/addresses.geoparquet")
md = pf.metadata
custom_meta = md.metadata
custom_meta[b"geo"] = json.dumps(geo_meta).encode("utf-8")

# Need to recreate with correct metadata. Write to temp, then use write_metadata approach.
table_with_meta = table.replace_schema_metadata(custom_meta)
pq.write_table(table_with_meta, "/work/addresses.geoparquet", version="2.6")

print("✅ addresses.geoparquet written successfully")
print(f"   Row count: {len(gdf)}")
print(f"   CRS: {gdf.crs}")

# Quick verification
df_check = gpd.read_parquet("/work/addresses.geoparquet")
print(f"   Verified: {len(df_check)} rows, CRS={df_check.crs}")
print(f"   dtypes:\n{df_check.dtypes}")

# Test the user's intended queries
mask = df_check["recorded_at"] > pd.Timestamp("2024-01-01")
print(f"   WHERE recorded_at > '2024-01-01': {mask.sum()} rows")
print(f"   SUM(unit_count): {df_check['unit_count'].sum()}")

import pandas as pd
import geopandas as gpd
from shapely import wkt

# Read CSV
df = pd.read_csv("/work/nyc_addresses.csv")

# Parse WKT geometry
df["geometry"] = df["geometry_wkt"].apply(wkt.loads)
gdf = gpd.GeoDataFrame(df, geometry="geometry", crs="EPSG:4326")
gdf.drop(columns=["geometry_wkt"], inplace=True)

# Convert recorded_at to datetime
gdf["recorded_at"] = pd.to_datetime(gdf["recorded_at"])

# Convert unit_count to int32
gdf["unit_count"] = gdf["unit_count"].astype("int32")

# All other columns as string
string_cols = ["id", "country", "postcode", "street", "number", "unit", "postal_city"]
for col in string_cols:
    gdf[col] = gdf[col].astype(str)

# Write GeoParquet — GeoPandas auto-generates the geo metadata.
# Set index=False so no extra __index_level_0__ column.
# Use version 2.6 Parquet for best timestamp support.
gdf.to_parquet("/work/addresses.geoparquet", index=False)

print("✅ addresses.geoparquet written")
print(f"   Rows: {len(gdf)}")
print(f"   CRS: {gdf.crs}")

# Verification
df_check = gpd.read_parquet("/work/addresses.geoparquet")
print(f"   Verified: {len(df_check)} rows, CRS={df_check.crs}")
print(f"   dtypes:\n{df_check.dtypes}")

# Test queries
mask = df_check["recorded_at"] > pd.Timestamp("2024-01-01")
print(f"   WHERE recorded_at > '2024-01-01': {mask.sum()} rows")
print(f"   SUM(unit_count): {df_check['unit_count'].sum()}")

import pandas as pd
import geopandas as gpd
from shapely import wkt

# Read CSV
df = pd.read_csv("/work/nyc_addresses.csv")

# Parse WKT geometry
df["geometry"] = df["geometry_wkt"].apply(wkt.loads)
gdf = gpd.GeoDataFrame(df, geometry="geometry", crs="EPSG:4326")
gdf.drop(columns=["geometry_wkt"], inplace=True)

# Convert recorded_at to datetime (stripping timezone -> tz-naive timestamp[us])
gdf["recorded_at"] = pd.to_datetime(gdf["recorded_at"]).dt.tz_localize(None)

# Convert unit_count to int32
gdf["unit_count"] = gdf["unit_count"].astype("int32")

# All other columns as string
string_cols = ["id", "country", "postcode", "street", "number", "unit", "postal_city"]
for col in string_cols:
    gdf[col] = gdf[col].astype(str)

# Write GeoParquet — GeoPandas auto-generates the geo metadata.
gdf.to_parquet("/work/addresses.geoparquet", index=False)

print("✅ addresses.geoparquet written")
print(f"   Rows: {len(gdf)}")
print(f"   CRS: {gdf.crs}")

# Verification
df_check = gpd.read_parquet("/work/addresses.geoparquet")
print(f"   Verified: {len(df_check)} rows, CRS={df_check.crs}")
print(f"   dtypes:\n{df_check.dtypes}")

# Test queries — use tz-naive comparison
mask = df_check["recorded_at"] > pd.Timestamp("2024-01-01")
print(f"   WHERE recorded_at > '2024-01-01': {mask.sum()} rows")
print(f"   SUM(unit_count): {df_check['unit_count'].sum()}")