fio-l1-nyc-csvwkt-addresses

Rerun

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

# ── 1. Read CSV ─────────────────────────────────────────────
df = pd.read_csv("/work/nyc_addresses.csv", dtype=str, keep_default_na=False)

# ── 2. Build geometry from WKT ──────────────────────────────
geometry = df["geometry_wkt"].apply(wkt.loads)
crs = "EPSG:4326"

# ── 3. Drop the WKT column ──────────────────────────────────
df = df.drop(columns=["geometry_wkt"])

# ── 4. Cast types ───────────────────────────────────────────
df["unit_count"] = df["unit_count"].astype("int32")
df["recorded_at"] = pd.to_datetime(df["recorded_at"])

# ── 5. Build GeoDataFrame ───────────────────────────────────
gdf = gpd.GeoDataFrame(df, geometry=geometry, crs=crs)

# ── 6. Build explicit PyArrow schema ────────────────────────
# All columns except recorded_at and unit_count stay as string (utf8).
# recorded_at → timestamp[us], unit_count → int32, geometry → binary WKB.
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 column will be added automatically by geopandas
])

# ── 7. Write GeoParquet ─────────────────────────────────────
gdf.to_parquet(
    "/work/addresses.geoparquet",
    schema=schema,
    index=False,
    engine="pyarrow",
)

print("✅ addresses.geoparquet written successfully")

# ── 8. Quick verification ───────────────────────────────────
verify = gpd.read_parquet("/work/addresses.geoparquet")
print(f"Rows: {len(verify)}")
print(f"CRS: {verify.crs}")
print(f"dtypes:\n{verify.dtypes}")
print(f"recorded_at sample: {verify['recorded_at'].head(3).values}")
print(f"unit_count min/max: {verify['unit_count'].min()} / {verify['unit_count'].max()}")
print(f"Filter test (recorded_at > '2024-01-01'): {len(verify[verify['recorded_at'] > pd.Timestamp('2024-01-01')])} rows")
print(f"SUM(unit_count): {verify['unit_count'].sum()}")

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

# ── 1. Read CSV ─────────────────────────────────────────────
df = pd.read_csv("/work/nyc_addresses.csv", dtype=str, keep_default_na=False)

# ── 2. Build geometry from WKT ──────────────────────────────
geometry = df["geometry_wkt"].apply(wkt.loads)

# ── 3. Drop the WKT column ──────────────────────────────────
df = df.drop(columns=["geometry_wkt"])

# ── 4. Cast types ───────────────────────────────────────────
df["unit_count"] = df["unit_count"].astype("int32")
df["recorded_at"] = pd.to_datetime(df["recorded_at"])

# ── 5. Build GeoDataFrame (needed for CRS-aware WKB encoding) ──
gdf = gpd.GeoDataFrame(df, geometry=geometry, crs="EPSG:4326")

# ── 6. Convert each column to a PyArrow array with desired type ──
def col_arr(name):
    if name == "recorded_at":
        return pa.Array.from_pandas(gdf[name], type=pa.timestamp("us"))
    elif name == "unit_count":
        return pa.array(gdf[name].values, type=pa.int32())
    elif name == "geometry":
        # Encode as WKB in EPSG:4326
        return pa.array(gdf.geometry.to_wkb().values, type=pa.binary())
    else:
        return pa.array(gdf[name].values, type=pa.utf8())

# ── 7. Build schema explicitly ──────────────────────────────
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()),
])

# ── 8. Build pyarrow Table with correct schema ──────────────
columns = [col_arr(name) for name in schema.names]
table = pa.table(columns, schema=schema)

# ── 9. Write as GeoParquet with geo metadata ────────────────
from geopandas.io.arrow import _create_metadata
geo_meta = _create_metadata(gdf)
pq.write_table(
    table,
    "/work/addresses.geoparquet",
    metadata_collector=None,
)
# Re-read to attach geo metadata properly
meta = table.schema.metadata or {}
if b"geo" not in meta:
    meta = {**meta, b"geo": geo_meta.serialize().to_pybytes()}
    table = table.replace_schema_metadata(meta)
pq.write_table(table, "/work/addresses.geoparquet")

print("✅ addresses.geoparquet written successfully")

# ── 10. Quick verification ──────────────────────────────────
verify = gpd.read_parquet("/work/addresses.geoparquet")
print(f"Rows: {len(verify)}")
print(f"CRS: {verify.crs}")
print(f"dtypes:\n{verify.dtypes}")
print(f"recorded_at sample: {verify['recorded_at'].head(3).values}")
print(f"unit_count min/max: {verify['unit_count'].min()} / {verify['unit_count'].max()}")

# Verify the filter works without CAST
filtered = verify[verify['recorded_at'] > pd.Timestamp('2024-01-01')]
print(f"Filter test (recorded_at > '2024-01-01'): {len(filtered)} rows")

# Verify SUM works
print(f"SUM(unit_count): {verify['unit_count'].sum()}")

# Verify DuckDB can read it cleanly
import duckdb
con = duckdb.connect()
con.execute("INSTALL spatial; LOAD spatial;")
result = con.execute("""
    SELECT COUNT(*) AS cnt, SUM(unit_count) AS total
    FROM '/work/addresses.geoparquet'
    WHERE recorded_at > '2024-01-01'
""").fetchone()
print(f"DuckDB verification: COUNT={result[0]}, SUM(unit_count)={result[1]}")

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

# ── 1. Read CSV ─────────────────────────────────────────────
df = pd.read_csv("/work/nyc_addresses.csv", dtype=str, keep_default_na=False)

# ── 2. Build geometry from WKT ──────────────────────────────
geometry = df["geometry_wkt"].apply(wkt.loads)

# ── 3. Drop the WKT column ──────────────────────────────────
df = df.drop(columns=["geometry_wkt"])

# ── 4. Cast types ───────────────────────────────────────────
df["unit_count"] = df["unit_count"].astype("int32")
df["recorded_at"] = pd.to_datetime(df["recorded_at"])

# ── 5. Build GeoDataFrame ───────────────────────────────────
gdf = gpd.GeoDataFrame(df, geometry=geometry, crs="EPSG:4326")

# ── 6. Build PyArrow table with explicit types ──────────────
arrays = []
field_types = {
    "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(),
}
fields = []
for col, typ in field_types.items():
    fields.append(pa.field(col, typ))
    arr = pa.array(gdf[col], type=typ)
    arrays.append(arr)

# Geometry column as WKB binary
geo_col = "geometry"
fields.append(pa.field(geo_col, pa.binary()))
arrays.append(pa.array(gdf.geometry.to_wkb(), type=pa.binary()))

schema = pa.schema(fields)

# ── 7. Create table ─────────────────────────────────────────
table = pa.table(arrays, schema=schema)

# ── 8. Attach GeoParquet metadata ───────────────────────────
from geopandas.io.arrow import _create_metadata
geo_meta = _create_metadata(gdf)
meta = table.schema.metadata or {}
meta[b"geo"] = geo_meta.serialize().to_pybytes()
table = table.replace_schema_metadata(meta)

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

print("✅ addresses.geoparquet written")

# ── 10. Verify ──────────────────────────────────────────────
verify = gpd.read_parquet("/work/addresses.geoparquet")
print(f"Rows: {len(verify)}, CRS: {verify.crs}")
print(f"recorded_at dtype: {verify['recorded_at'].dtype}")
print(f"unit_count dtype:  {verify['unit_count'].dtype}")
filtered = verify[verify['recorded_at'] > pd.Timestamp('2024-01-01')]
print(f"Filter test (recorded_at > '2024-01-01'): {len(filtered)} rows")
print(f"SUM(unit_count): {verify['unit_count'].sum()}")

# DuckDB verification
import duckdb
con = duckdb.connect()
con.execute("INSTALL spatial; LOAD spatial;")
row = con.execute("""
    SELECT COUNT(*) AS cnt, SUM(unit_count) AS total
    FROM '/work/addresses.geoparquet'
    WHERE recorded_at > '2024-01-01'
""").fetchone()
print(f"DuckDB:  COUNT={row[0]}, SUM(unit_count)={row[1]}")

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

# ── 1. Read CSV ─────────────────────────────────────────────
df = pd.read_csv("/work/nyc_addresses.csv", dtype=str, keep_default_na=False)

# ── 2. Build geometry from WKT ──────────────────────────────
geometry = df["geometry_wkt"].apply(wkt.loads)

# ── 3. Drop the WKT column ──────────────────────────────────
df = df.drop(columns=["geometry_wkt"])

# ── 4. Cast types ───────────────────────────────────────────
df["unit_count"] = df["unit_count"].astype("int32")
df["recorded_at"] = pd.to_datetime(df["recorded_at"])

# ── 5. Build GeoDataFrame ───────────────────────────────────
gdf = gpd.GeoDataFrame(df, geometry=geometry, crs="EPSG:4326")

# ── 6. Build PyArrow table with explicit types ──────────────
field_types = {
    "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(),
}
fields = []
arrays = []
for col, typ in field_types.items():
    fields.append(pa.field(col, typ))
    arrays.append(pa.array(gdf[col], type=typ))

# Geometry column as WKB binary
fields.append(pa.field("geometry", pa.binary()))
arrays.append(pa.array(gdf.geometry.to_wkb(), type=pa.binary()))

schema = pa.schema(fields)
table = pa.table(arrays, schema=schema)

# ── 7. Attach GeoParquet metadata ───────────────────────────
from geopandas.io.arrow import _create_metadata
geo_meta = _create_metadata(gdf)  # dict

meta = table.schema.metadata or {}
meta[b"geo"] = json.dumps(geo_meta).encode("utf-8")
table = table.replace_schema_metadata(meta)

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

print("✅ addresses.geoparquet written")

# ── 9. Verify ──────────────────────────────────────────────
verify = gpd.read_parquet("/work/addresses.geoparquet")
print(f"Rows: {len(verify)}, CRS: {verify.crs}")
print(f"recorded_at dtype: {verify['recorded_at'].dtype}")
print(f"unit_count dtype:  {verify['unit_count'].dtype}")
filtered = verify[verify['recorded_at'] > pd.Timestamp('2024-01-01')]
print(f"Filter test (recorded_at > '2024-01-01'): {len(filtered)} rows")
print(f"SUM(unit_count): {verify['unit_count'].sum()}")

# DuckDB verification
import duckdb
con = duckdb.connect()
con.execute("INSTALL spatial; LOAD spatial;")
row = con.execute("""
    SELECT COUNT(*) AS cnt, SUM(unit_count) AS total
    FROM '/work/addresses.geoparquet'
    WHERE recorded_at > '2024-01-01'
""").fetchone()
print(f"DuckDB:  COUNT={row[0]}, SUM(unit_count)={row[1]}")