fio-l1-nyc-csvwkt-addresses

Rerun

import pandas as pd
import geopandas as gpd
from shapely import wkt
from pyarrow import Table, Schema, field, timestamp, int32, string
import pyarrow.parquet as pq

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

# Convert geometry from WKT to shapely geometry (EPSG:4326)
df['geometry'] = df['geometry_wkt'].apply(wkt.loads)

# Determine CRS (EPSG:4326)
crs = "EPSG:4326"

# Create GeoDataFrame
gdf = gpd.GeoDataFrame(df, geometry='geometry', crs=crs)

# Drop the geometry_wkt column
gdf = gdf.drop(columns=['geometry_wkt'])

# Convert types
gdf['recorded_at'] = pd.to_datetime(gdf['recorded_at'])
gdf['unit_count'] = gdf['unit_count'].astype('int32')

# Ensure all other columns are string (fill NaN with empty string)
string_cols = ['id', 'country', 'postcode', 'street', 'number', 'unit', 'postal_city']
for col in string_cols:
    gdf[col] = gdf[col].fillna('').astype(str)

# Write to GeoParquet using pyarrow with explicit schema
# First, convert geometry to WKB
import pyarrow as pa

# Prepare geometry column as WKB binary
from shapely import to_wkb
geometry_wkb = to_wkb(gdf.geometry.values)

# Build the schema
schema = Schema([
    field('id', string()),
    field('country', string()),
    field('postcode', string()),
    field('street', string()),
    field('number', string()),
    field('unit', string()),
    field('postal_city', string()),
    field('recorded_at', timestamp('us')),
    field('unit_count', int32()),
])

# Build table
table = Table.from_pydict({
    'id': gdf['id'].values,
    'country': gdf['country'].values,
    'postcode': gdf['postcode'].values,
    'street': gdf['street'].values,
    'number': gdf['number'].values,
    'unit': gdf['unit'].values,
    'postal_city': gdf['postal_city'].values,
    'recorded_at': gdf['recorded_at'].values,
    'unit_count': gdf['unit_count'].values,
}, schema=schema)

# Write GeoParquet with geo metadata
import json
geo_metadata = {
    'columns': {
        'geometry': {
            'encoding': 'WKB',
            'geometry_types': ['Point'],
            'crs': {
                'type': 'GeographicCRS',
                'name': 'WGS 84',
                'datum': {
                    'type': 'GeodeticReferenceFrame',
                    'name': 'World Geodetic System 1984',
                },
                'coordinate_system': {
                    'axis': [
                        {'name': 'Latitude', 'abbreviation': 'lat', 'direction': 'north', 'unit': 'degree'},
                        {'name': 'Longitude', 'abbreviation': 'lon', 'direction': 'east', 'unit': 'degree'},
                    ]
                },
            },
        }
    },
    'primary_column': 'geometry',
}

# Write using pyarrow parquet writer with geo metadata
pq.write_table(table, '/work/addresses.geoparquet', 
               compression='snappy',
               metadata_collector=None)

# Now manually add the geo metadata using ParquetFile
# Actually, let's use geopandas' to_parquet which handles geo metadata properly

print("File written. Now let's verify...")

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

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

# Convert geometry from WKT to shapely geometry
df['geometry'] = df['geometry_wkt'].apply(wkt.loads)

# Create GeoDataFrame with EPSG:4326
gdf = gpd.GeoDataFrame(df, geometry='geometry', crs="EPSG:4326")

# Drop the geometry_wkt column
gdf = gdf.drop(columns=['geometry_wkt'])

# Convert types
gdf['recorded_at'] = pd.to_datetime(gdf['recorded_at'])
gdf['unit_count'] = gdf['unit_count'].astype('int32')

# Ensure all other columns are string (fill NaN with empty string)
string_cols = ['id', 'country', 'postcode', 'street', 'number', 'unit', 'postal_city']
for col in string_cols:
    gdf[col] = gdf[col].fillna('').astype(str)

# Write to GeoParquet using GeoPandas
# Use geoarrow/geo metadata via the parquet engine
gdf.to_parquet('/work/addresses.geoparquet', index=False, compression='snappy')

print("Done. Verifying...")

# Verify
import pyarrow.parquet as pq
pf = pq.ParquetFile('/work/addresses.geoparquet')
print("Schema:")
print(pf.schema)
print("\nGeo metadata:")
print(pf.schema.metadata)

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

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

# Convert geometry from WKT to shapely geometry
df['geometry'] = df['geometry_wkt'].apply(wkt.loads)

# Create GeoDataFrame with EPSG:4326
gdf = gpd.GeoDataFrame(df, geometry='geometry', crs="EPSG:4326")

# Drop the geometry_wkt column
gdf = gdf.drop(columns=['geometry_wkt'])

# Convert types
gdf['recorded_at'] = pd.to_datetime(gdf['recorded_at'])
gdf['unit_count'] = gdf['unit_count'].astype('int32')

# Ensure all other columns are string (fill NaN with empty string)
string_cols = ['id', 'country', 'postcode', 'street', 'number', 'unit', 'postal_city']
for col in string_cols:
    gdf[col] = gdf[col].fillna('').astype(str)

# Write to GeoParquet using GeoPandas
gdf.to_parquet('/work/addresses.geoparquet', index=False, compression='snappy')

print("Done. Verifying...")

import pyarrow.parquet as pq
pf = pq.ParquetFile('/work/addresses.geoparquet')
print("Schema:")
print(pf.schema)
print("\nMetadata:")
meta = pf.metadata
print(meta)
if meta.metadata:
    print("\nKey metadata items:")
    for k, v in meta.metadata.items():
        print(f"  {k}: {v[:200]}...")
else:
    print("No metadata found")

import duckdb

# Test queries without CASTs
con = duckdb.connect()

# Test 1: WHERE recorded_at > '2024-01-01'
result1 = con.execute("""
    SELECT COUNT(*) as cnt 
    FROM '/work/addresses.geoparquet' 
    WHERE recorded_at > '2024-01-01'
""").fetchone()
print(f"Rows with recorded_at > '2024-01-01': {result1[0]}")

# Test 2: SUM(unit_count)
result2 = con.execute("""
    SELECT SUM(unit_count) as total_units 
    FROM '/work/addresses.geoparquet'
""").fetchone()
print(f"Total unit_count: {result2[0]}")

# Test 3: Combined
result3 = con.execute("""
    SELECT SUM(unit_count) as total_units 
    FROM '/work/addresses.geoparquet' 
    WHERE recorded_at > '2024-06-01'
""").fetchone()
print(f"Total unit_count after 2024-06-01: {result3[0]}")

# Also test with duckdb spatial extension for geometry
con.execute("INSTALL spatial; LOAD spatial;")
result4 = con.execute("""
    SELECT ST_GeometryType(geometry) as geom_type, 
           COUNT(*) as cnt
    FROM '/work/addresses.geoparquet'
    GROUP BY geom_type
""").fetchone()
print(f"Geometry type: {result4[0]}, count: {result4[1]}")

# Verify CRS
result5 = con.execute("""
    SELECT DISTINCT ST_SRID(geometry) as srid
    FROM '/work/addresses.geoparquet'
    LIMIT 1
""").fetchone()
print(f"SRID: {result5[0] if result5 else 'N/A'}")

con.close()
print("\nAll queries work without CASTs!")

import duckdb

con = duckdb.connect()
con.execute("INSTALL spatial; LOAD spatial;")

# Test 1: WHERE recorded_at > '2024-01-01'
result1 = con.execute("""
    SELECT COUNT(*) as cnt 
    FROM ST_Read('/work/addresses.geoparquet') 
    WHERE recorded_at > '2024-01-01'
""").fetchone()
print(f"Rows with recorded_at > '2024-01-01': {result1[0]}")

# Test 2: SUM(unit_count)
result2 = con.execute("""
    SELECT SUM(unit_count) as total_units 
    FROM ST_Read('/work/addresses.geoparquet')
""").fetchone()
print(f"Total unit_count: {result2[0]}")

# Test 3: Combined
result3 = con.execute("""
    SELECT SUM(unit_count) as total_units 
    FROM ST_Read('/work/addresses.geoparquet') 
    WHERE recorded_at > '2024-06-01'
""").fetchone()
print(f"Total unit_count after 2024-06-01: {result3[0]}")

# Check geometry type
result4 = con.execute("""
    SELECT ST_GeometryType(geometry) as geom_type, 
           COUNT(*) as cnt
    FROM ST_Read('/work/addresses.geoparquet')
    GROUP BY geom_type
""").fetchone()
print(f"Geometry type: {result4[0]}, count: {result4[1]}")

# Verify CRS
result5 = con.execute("""
    SELECT DISTINCT ST_SRID(geometry) as srid
    FROM ST_Read('/work/addresses.geoparquet')
    LIMIT 1
""").fetchone()
print(f"SRID: {result5[0] if result5 else 'N/A'}")

con.close()
print("\nAll queries work without CASTs!")

import duckdb

con = duckdb.connect()

# Test reading as parquet directly (DuckDB has native parquet support)
# Test 1: WHERE recorded_at > '2024-01-01'
result1 = con.execute("""
    SELECT COUNT(*) as cnt 
    FROM read_parquet('/work/addresses.geoparquet') 
    WHERE recorded_at > '2024-01-01'
""").fetchone()
print(f"Rows with recorded_at > '2024-01-01': {result1[0]}")

# Test 2: SUM(unit_count)
result2 = con.execute("""
    SELECT SUM(unit_count) as total_units 
    FROM read_parquet('/work/addresses.geoparquet')
""").fetchone()
print(f"Total unit_count: {result2[0]}")

# Test 3: Combined
result3 = con.execute("""
    SELECT SUM(unit_count) as total_units 
    FROM read_parquet('/work/addresses.geoparquet') 
    WHERE recorded_at > '2024-06-01'
""").fetchone()
print(f"Total unit_count after 2024-06-01: {result3[0]}")

# Check the actual types
result4 = con.execute("""
    DESCRIBE SELECT * FROM read_parquet('/work/addresses.geoparquet')
""").fetchall()
print("\nColumn types:")
for row in result4:
    print(f"  {row[0]}: {row[1]}")

con.close()
print("\nAll queries work without CASTs!")