ETL Vzor: Práce s dobře známými soubory v cloudu

bharold

Už nějakou dobu se data založená na souborech přesouvají na web a jsme zvyklí ručně stahovat, co potřebujeme, a dále s tím pracovat v našem desktopovém softwaru jako ArcGIS Pro.<\/P>

Tento blog je o automatizaci<\/STRONG> tohoto procesu, nahrazení kroků stahování a zpracování snadno pochopitelnými, dobře definovanými a sdílitelnými nástroji, které dodávají váš informační produkt na vyžádání<\/STRONG>.<\/P>Na dvou koncích škály dat mám celosvětové politické dělení a činnost stavebních povolení ve Vancouveru v Kanadě.  První dataset je velký<\/STRONG> a pochází z AWS S3 bucketu, druhý je malý<\/STRONG> a pochází z portálu otevřených dat.  Oba jsou načítány pomocí notebooků v ArcGIS Pro se stejnými základními nástroji.<\/STRONG><\/P>Nejprve malý dataset:<\/P>Stavební povolení ve Vancouveru<\/span><\/span><\/P>Velký dataset:<\/P>Celosvětové politické dělení<\/span><\/span><\/P>A uprostřed, politické dělení pouze pro Německo:<\/P>Pohled na politické dělení Německa<\/span><\/span><\/P>Všechny tyto datasety byly načteny podobně, ale s pozoruhodnými rozdíly, které si níže rozebereme.<\/P>Pokud neznáte GeoParquet<\/STRONG><\/A>, brzy ho poznáte, a dělám z něj svůj dobře známý formát<\/STRONG> pro tento příspěvek.  Další běžné cloudové typy souborů jako CSV, Excel a JSON by také fungovaly.  Pod pojmem "cloud" nemyslím jen místo, kde je získáváte, ale formáty umožňující vzdálené dotazování bez nutnosti stahovat a kontrolovat lokální kopii.<\/P>I když říkám, že stejné základní nástroje mohou přistupovat k velkým datům na S3 i malým datům na webu<\/STRONG>, záleží na schopnostech příslušných serverů.  S3 umí doručovat soubory podle dotazů proti hive storage, zatímco webová stránka umí jen doručit stažení celých souborů.<\/P>Pojďme nejprve vyřešit menší dataset - stavební povolení, automatickým stažením ze stránky.<\/P>Data jsou umístěna v portálu otevřených dat<\/STRONG><\/A>, pokud procházíte stránku, uvidíte možnosti formátu ke stažení.<\/P>Data jsou automatizována v notebooku VancouverBuildingPermits<\/STRONG>.  Pokud si prohlédnete kód, všimnete si, že používám DuckDB<\/STRONG><\/A> - je to prostorově podporovaná SQL databáze a webový klient.  Uvidíte, že stahování dat a převod do feature class je velmi jednoduchý - nástroje odvozují schéma ze zdroje Parquet plus SQL příkazy<\/STRONG>.<\/P> <\/P>import arcpy from arcgis.features import GeoAccessor, GeoSeriesAccessor import duckdb import math import os arcpy.env.overwriteOutput = True url = r"https:\/\/opendata.vancouver.ca\/api\/explore\/v2.1\/catalog\/datasets\/issued-building-permits\/exports\/parquet?lang=en&refine.issuedate=2024&timezone=America%2FLos_Angeles" conn = duckdb.connect() conn.sql("set force_download=true;") conn.sql("install httpfs;load httpfs;") conn.sql("install spatial;load spatial;") conn.sql(f"create temp view permitsView as select * from read_parquet('{url}');") bldgPermits = conn.sql("""select permitnumber as Permit_Number, try_cast (permitnumbercreateddate as date) as Permit_Number_Created_Date, try_cast (issuedate as date) as Issue_Date, permitelapseddays::int as Permit_Elapsed_Days, projectvalue::float as Project_Value, typeofwork as Type_Of_Work, address as Address, projectdescription as Project_Description, permitcategory as Permit_Category, applicant as Applicant, applicantaddress as Applicant_Address, propertyuse as Property_Use, specificusecategory as Specific_Use_Category, buildingcontractor as Building_Contractor, buildingcontractoraddress as Building_Contractor_Address, issueyear as Issue_Year, geolocalarea as Geo_Loc_Area, yearmonth as Year_Month, ST_AsText(ST_GeomFromWKB(geom)) as SHAPE from permitsView;""") df = bldgPermits.df() df.spatial.set_geometry("SHAPE",sr=4326,inplace=True) aprx = arcpy.mp.ArcGISProject("CURRENT") gdb = aprx.defaultGeodatabase out_fc = "Issued_Building_Permits" location = os.path.join(gdb,out_fc) df.spatial.to_featureclass(location,sanitize_columns=False)<\/code><\/pre><P>&nbsp;<\/P><P>Pokud si prohlédnete kód ve staženém souboru, uvidíte některé další kroky, které jsem použil k zaokrouhlení šířek textových polí na násobky 10, což je zde zbytečný detail, ale ukazuje to, jak to udělat, pokud očekáváte budoucí úpravy dat a nechcete ořezávání hodnot.<\/P><P>Toto je tedy jednoduchý případ načtení malého cloudového souboru do ArcGIS s minimální kontrolou nad schématem - opět si všimněte, že jsou k dispozici standardní SQL příkazy pro definici vašeho pohledu na cloudový soubor.&nbsp; Porovnejte tento přístup s definicí schématu pomocí geoprocessingu v <A title="Refining the schema for local well-known files" href="https:\/\/community.esri.com\/"><STRONG>tomto souvisejícím příspěvku<\/STRONG><\/A>.<\/P><P>Přejděme k datům globálního rozsahu pomocí notebooku&nbsp;<STRONG>Division_Area<\/STRONG>, využívající data z <A title="Overture" href="https:\/\/overturemaps.org\/"><STRONG>Overture Maps Foundation<\/STRONG><\/A>.<\/P><P>&nbsp;<\/P><pre class="lia-code-sample language-python"><code># Získání vrstvy Division_Area z Overture Maps Foundation import arcpy from datetime import datetime import duckdb import os aprx = arcpy.mp.ArcGISProject("CURRENT") homeFolder = aprx.homeFolder gdb = aprx.defaultGeodatabase release = "2024-07-22.0" def getNow(): return str(datetime.utcnow().replace(microsecond=0)) arcpy.env.overwriteOutput = True sR = arcpy.SpatialReference(4326) conn = duckdb.connect() conn.sql("install spatial;load spatial;") conn.sql("install httpfs;load httpfs;") conn.sql("set s3_region='us-west-2';") conn.sql("set enable_object_cache=true;") # Vytvoření DuckDB relace # Vložte libovolnou požadovanou podmínku where, například "country = 'GB'" # Výraz order-by pomáhá zobrazit malé divize nad velkými, ale je nákladný print(f"Zpracování začíná v {getNow()}") whereExp = "1 = 1" sql = f"""select id, bbox.xmin as xmin, bbox.ymin as ymin, bbox.xmax as xmax, bbox.ymax as ymax, version, subtype, names.primary as primary_name, class, region, country, norms.driving_side as driving_side, geometry from read_parquet('s3:\/\/_overturemaps-us-west-2/release/{release}/theme=divisions/type=division_area/*',filename=true, hive_partitioning=1) where {whereExp} order by ST_Area(ST_GeomFromWKB(geometry)) desc;""" duckDivisions = conn.sql(sql) # Vytvoření výstupní feature class s přesnou kontrolou schématu print('Vytvářím výstupní feature class v {}'.format(getNow())) arcDivisions = arcpy.management.CreateFeatureclass(out_path=gdb, out_name="Division_Areas", geometry_type="POLYGON", spatial_reference=sR).getOutput(0) arcpy.management.AddField(in_table=arcDivisions,field_name="id",field_type="TEXT",field_length=32) for f in ["xmin","ymin","xmax","ymax"]: arcpy.management.AddField(in_table=arcDivisions,field_name=f,field_type="FLOAT") arcpy.management.AddField(in_table=arcDivisions,field_name="version",field_type="SHORT") arcpy.management.AddField(in_table=arcDivisions,field_name="subtype",field_type="TEXT",field_length=20) arcpy.management.AddField(in_table=arcDivisions,field_name="primary_name",field_type="TEXT",field_length=100) arcpy.management.AddField(in_table=arcDivisions,field_name="class",field_type="TEXT",field_length=6) arcpy.management.AddField(in_table=arcDivisions,field_name="region",field_type="TEXT",field_length=10) arcpy.management.AddField(in_table=arcDivisions,field_name="country",field_type="TEXT",field_length=2) arcpy.management.AddField(in_table=arcDivisions,field_name="driving_side",field_type="TEXT",field_length=10) # Napište výstup print('Zápis výstupu v {}'.format(getNow())) s arcpy.da.InsertCursor(arcDivisions,["id","xmin","ymin","xmax","ymax","version", "subtype","primary_name","class","region", "country","driving_side","shape@"]) jako iCursor: row = duckDivisions.fetchone() i = 1 zatímco row: pokud i % 100000 == 0: print(f"Vloženo {i} řádků Division_Areas...") row = list(row) row[-1] = arcpy.FromWKB(row[-1]) iCursor.insertRow(row) i+=1 row = duckDivisions.fetchone() del iCursor # Oprava geometrií, odstranění nulových geometrií print(f"Oprava špatných geometrií v {getNow()}") arcpy.management.RepairGeometry( in_features=arcDivisions, delete_null="DELETE_NULL", validation_method="OGC") print(f"Dokončeno v {getNow()}") 

Hlavní rozdíly mezi tímto globálním načítáním dat a načítáním pro stavební povolení jsou, že zdroj je získán pomocí wildcard dotazu na S3 hive storage, a schéma je definováno ručně.  Nástroje jsou stejné.< /P>

Takže to je pohled na uživatelsky orientované načítání dat ve velkém i malém měřítku, ale co když jste vlastník dat a chcete nabídnout nekonečně flexibilní cloud well-known file-based dodávku dat širokému okruhu uživatelů, aniž by museli řešit tvorbu nástrojů?  Tady přichází příklad středního měřítka politických divizí pro Německo.< /P>

Knihovna klienta, kterou používáme - DuckDB - je také formát úložiště, ale s prostorovými databázovými schopnostmi, včetně pohledů, takže v případě dat ze cloudu je jednoduché nabídnout datový produkt, který obsahuje pouze pohled, který je čten na vyžádání jakýmkoli koncovým uživatelem.  To znamená, že za úsilí definování pohledu lze datový produkt jakéhokoli rozsahu dodat v malém databázovém souboru.< /P>

Blog ke stažení obsahuje takový datový produkt - mydivisionview.duckdb - který byl vytvořen notebookem MakeViewDatabase a může být načten kýmkoli pomocí notebooku ReadViewDatabase.  V tomto případě může soubor o velikosti pouhých 268KB dodat data národní (nebo dokonce globální) úrovně.  Toto je síla cloud-native well-known files čtených na vyžádání z cloudu.  Pokud se zdroj datasetu změní, stejná databáze DuckDB vždy přistoupí k nejnovějším datům.< /P>

Níže je kód pro ReadViewDatabase.  Pokud zkopírujete mydivisionview.duck db do složky vašeho projektu, dodá obsah pohledu do výchozí geodatabáze vašeho projektu během sekund (15 na mém počítači):

import arcpy
from datetime import datetime
import duckdb
import os
def getNow():
    return str(datetime.utcnow().replace(microsecond=0))
print(f"Extrahování dat v {getNow()}")
arcpy.env.overwriteOutput = True
aprx = arcpy.mp.ArcGISProject("CURRENT")
homeFolder = aprx.homeFolder
gdb = aprx.defaultGeodatabase
sR = arcpy.SpatialReference(4326)
duckDB = os.path.join(homeFolder,"MyDivisionView.duckdb")
conn = duckdb.connect(duckDB)
conn.sql("install spatial;load spatial;")
conn.sql("install httpfs;load httpfs;")
conn.sql("set enable_object_cache=true;")
duckDivisions = conn.sql(f"""select id,
                                xmin::float as xmin,
                                ymin::float as ymin,
                                xmax::float as xmax,
                                ymax::float as ymax,
                                version::short as version,
                                subtype,
                                primary_name,
                                class,
                                region,
                                country,
                                driving_side,
                                geometry
                                from MyDivisionView;""")
arcDivisions = arcpy.management.CreateFeatureclass(gdb,
                                                   out_name="My_Division_Areas",
                                                   geometry_type="POLYGON",
                                                   spatial_reference=sR).getOutput(0)
arcpy.management.AddField(in_table=arcDivisions,field_name="id",field_type="TEXT",field_length=32)
for f in ["xmin","ymin","xmax","ymax"]:
    arcpy.management.AddField(in_table=arcDivisions,field_name=f,field_type="FLOAT")
arcpy.management.AddField(in_table=arcDivisions,field_name="version",field_type="SHORT")
arcpy.management.AddField(in_table=arcDivisions,field_name="subtype",field_type="TEXT",field_length=20)
arcpy.management.AddField(in_table=arcDivisions,field_name="primary_name",field_type="TEXT",field_length=100)
arcpy.management.AddField(in_table=arcDivisions,field_name="class",field_type="TEXT",field_length=6)
arcpy.management.AddField(in_table=arcDivisions,field_name="region",field_type="TEXT",field_length=10)
arcpy.management.AddField(in_table=arcDivisions,field_name="country",field_type="TEXT",field_length=2)
arcpy.management.AddField(in_table=arcDivisions,field_name="driving_side",field_type="TEXT",field_length=10)
s with arcpy.da.InsertCursor(arcDivisions,["id","xmin","ymin","xmax","ymax","version","subtype",
                                         "primary_name","class","region","country",
                                         "driving_side","shape@"]) jako iCursor:
    row = duckDivisions.fetchone()
    i = 1
    zatímco row:
        #if i % 1000 == 0:
        #     print('Inserted {} My_Division_Area rows'.format(i))
        row = list(row)
        row[-1] = arcpy.FromWKB(row[-1])
        iCursor.insertRow(row)
        i+=1
        row = duckDivisions.fetchone()
del iCursor
conn.close()
print(f"Data extrahována v {getNow()}")

Dokončím několika shrnutími:< /P>

• Malé cloud well-known files v cloudu se snadno načítají.< /LI>
• Neomezeně velké cloud well-known files v objektovém úložišti se snadno načítají.< /LI>
• Data cloud well-known files jakéhokoli rozsahu v jakékoli definici pohledu lze snadno sdílet v malých kontejnerech.< /LI>

 < /P>

 < /P>

 < /P>

CodyPatterson

Hey @BruceHarold 

Great post! I can definitely see myself using this in the future, we have extremely large datasets and your mention of geoparquet in very intuitive, lots of new names of software I haven't used before in here!

Thank you for sharing!

Cody