Kolega se mě nedávno zeptal, jak přesunout několik miliard záznamů do GeoEventova spatiotemporálního big data úložiště (STBDS) u zákazníka, pomocí ArcGIS Data Interoperability. Jedná se o archivované polohy vozidel pro energetickou společnost, uložené v Oracle geodatabázi. Denně přichází několik milionů nových událostí. Archivovaná data jsou v více než 70 tabulkách.<\/P>
Protože feature service ve STBDS má stejný REST API jako běžný hostovaný feature service, plán byl využít vzor popsaný v mém dřívějším blogu<\/A>, konkrétně ETL archivních dat do portal shapefile položek a pak použít Append REST endpoint cílové vrstvy k asynchronnímu načtení dat. Další komplikací bylo paralelizovat ETL jako více současných úloh. Tento přístup by snížil riziko výpadků sítě při streamování malých transakcí (1000 prvků je výchozí velikost dávky) do portálu. Ukázalo se, že zákaznické prostředí nebylo na verzi podporující tento workflow a šli jsme tedy cestou streamování, ale přimělo mě to přemýšlet, jak ETL velká data do nové generace cloudových skladů, které můžete používat nativně, a v ArcGIS Pro 2.9.<\/P>Moje základní sdělení je, že tyto lift and shift úlohy můžete dělat přesunem velkých datasetů jako souborů, ve více současných úlohách, čímž maximalizujete propustnost a minimalizujete riziko přenosu. Podívejme se jak.<\/STRONG><\/P>Cloudové sklady jako Snowflake, Big Query a Redshift lze dotazovat a číst v ArcGIS Pro 2.9, ale nelze do nich zapisovat pomocí standardních nástrojů. ArcGIS Data Interoperability však může do těchto skladů zapisovat, včetně prostorových dat, ale výchozí režim je streamování, které nemusí škálovat podle vašich potřeb. Ukážu vám vzor použitelný pro všechny tři sklady:<\/P>ETL celých datasetů, včetně prostorových dat, jako Apache Parquet nebo jiný formát jako CSVGeometrie je kódována ve znakově založeném standardním formátu<\/LI><\/UL><\/LI>Automatizujte ETL ve více současných procesech<\/STRONG><\/LI>Nepište žádný kód<\/STRONG> kromě SQL nebo makro příkazů požadovaných cílovým prostředím<\/LI><\/UL>Taky hodím kost pro každého programátora číhajícího v mém no-code blogovém prostoru, viz níže 😉<\/span>, tj. pár Python tipů na tvorbu Parquet souborů (Poznámka: Parquet soubory jsou podporovaným typem položky v ArcGIS Online od 22. září 2021. Sdílejte nějaké!)
<\/span>Miliardy prvků jsou v rozsahu tohoto vzoru, ale používám skromnější dataset pro demonstrační účely, pouze 2,3 milionu bodových prvků.<\/P>
2.3 Milionu bodových prvků<\/span><\/span><\/P>Moje data jsou ve 12 feature classes, můžete mít libovolný počet. Vzor který ukážu funguje s daty rozdělenými na samostatné části zpracovatelné současně. Pokud máte monolitická data, rozdělte je buď prostorově sami (
oriented fishnet anyone?<\/A>) nebo přidáním pole označujícího identifikátor dávky naplněného pozicí řádku - toto pole můžete během zpracování odstranit.<\/P>Zmínil jsem Snowflake, Big Query a Redshift sklady. Ve všech případech můžete umístit Parquet soubory tam, kde je cílové prostředí vidí a pak načíst z těchto Parquet souborů. Pro prostorová data musí být Parquet soubory s geometrií kódovanou ve formátu pochopitelném cílovým prostředím (
Snowflake<\/A> a Big Query<\/A> podporují GeoJSON a WKT, Redshift<\/A> podporuje WKT). Poskytnu pouze ukázkový příklad s GeoJSON pro Snowflake. Moje demo data jsou bodová geometrie a pole které používám pro uložení GeoJSON má šířku 100, pokud používáte polyline nebo polygonová data měli byste zjistit jak široké jsou vaše nejbohatší prvky při kódování pole. Například vybral jsem velmi bodově bohatý polygon ve vrstvě a jako GeoJSON má 2 071 156 znaků:<\/P> <\/P>
s arcpy.da.SearchCursor('NZ Property Titles','shape@') as cursor:
Poznámka:<\STRONG> Pro Big Query má Data Interoperability GoogleBigQueryConnector<\A>, hub transformer který může načítat CSV do tabulek. To může být jednodušší než posílání Parquet a použití bq příkazového prostředí k načtení dat, tento scénář jsem nezkoumal.<\P>Pojďme se ponořit do mého konkrétního workflow. Tajná ingredience je vytvořit dva<\STRONG> ETL nástroje, první řídí úlohy a volá
WorkspaceRunner<\A>, transformer který volá druhý nástroj provádějící práci. Je to velmi jednoduché, zde je LoadManager.fmw<\STRONG>, bere seznam argumentů, v mém případě názvy feature classes v geodatabázi:<\P>
LoadManager<\span><\span><\P> WorkspaceRunner spustí až 7 FME procesů které běží cílový nástroj dokud není fronta úloh vyčerpána. Zpracování bude pravděpodobně CPU bound zatímco pracovní procesy extrahují, kódují a nahrávají datasetové soubory. Každému procesu jsem dovolil spustit dvě úlohy což spotřebovalo mé příchozí datasety ve 6 procesech.<\P>
WorkspaceRunner<\span><\span><\P> <\P>Zde je LoadWorkerParquet.fmw<\STRONG>.<\P>
LoadWorkerParquet<\span><\span><\P>Také je to jednoduchý nástroj, čte geodatabázi, zapisuje lokální parquet soubor a pak posílá parquet soubor do Snowflake kde jsou data kopírována do tabulky. Nechám vás prohlédnout si SQLExecutor sami ale jako pomůcku k pochopení zde po substituci proměnných vypadá příkaz (pro Snowflake):<\P> <\P>create or replace file format Canterburyparquet_format
type = 'parquet';
create or replace temporary stage stageCanterbury
file_format = Canterburyparquet_format;
put file:\/C:\Work\Parquet\TitlesCanterbury.parquet @stageCanterbury;
copy into "INTEROPERABILITY"."PUBLIC"."Titles"
from (select
$1:id::number,
$1:title_no::varchar,
$1:status::varchar,
$1:type::varchar,
$1:land_district::varchar,
$1:issue_date::timestamp,
$1:guarantee_status::varchar,
$1:estate_description::varchar,
$1:number_owners::varchar,
$1:spatial_extents_shared::varchar,
to_geography($1:geom::varchar)
from @stageCanterbury);<\code><\pre> <\P>Až parquet soubor dorazí do Snowflake je ingest bleskurychlý. Mimochodem naučil jsem se k tomu udělat přečtením nápovědy, nejsem Snowflake DBA.<\/P>Jaký výkon byste měli očekávat? Při psaní jsem uvězněn doma jako ostatní, ale na své domácí WiFi načtu 2,3 milionu features do Snowflake za 6 minut, takže s dobrým počítačem a kabelovou sítí si myslím konzervativně 25 milionů point features za hodinu. Samozřejmě pro produkční prostředí a opravdu velký úkol můžete použít více počítačů, cílové cloudové sklady určitě škálují, aby zvládly propustnost.<\/P>V blogovém downloadu si všimnete, že zahrnuji druhý pracovní nástroj, LoadWorker.fmw, to bylo pro mě k porovnání výkonu obvyklého způsobu zápisu do Snowflake s 100K features na transakci, bylo to mnohem pomalejší.<\/P>Nyní zpět v core Pro 2.9 jsou moje data načtena do Snowflake a mohu na ně házet dotazy a užívat si škálovaný výpočetní zážitek.<\/P>
Snowflake in Catalog Pane<\/span><\/span><\/P>Zmínil jsem možnost Pythonu pro vytváření parquet souborů, je to v blogovém downloadu, ale tady je to také:<\/P> <\/P># Pro 2.9+ příklad vytvoření parquet souboru z feature class
# Geometrie je zakódována jako GeoJSON v poli 'geom'
import arcpy
import pyarrow.parquet as pq
arcpy.env.overwriteOutput = True
# Zdrojová feature class
Canterbury = r"C:\Work\Parquet\Parquet.gdb\Canterbury"
# Vytvořit in-memory feature class ve WGS84
with arcpy.EnvManager(outputCoordinateSystem='GEOGCS["GCS_WGS_1984",DATUM["D_WGS_1984",SPHEROID["WGS_1984",6378137.0,298.257223563]],PRIMEM["Greenwich",0.0],UNIT["Degree",0.0174532925199433]]',
geographicTransformations="NZGD_2000_To_WGS_1984_1"):
arcpy.conversion.ExportFeatures(
in_features=Canterbury,
out_features=r"memory\Canterbury")
# Přidat pole geom (geometrie jiná než point bude vyžadovat širší pole)
Canterbury = arcpy.management.AddField("memory\Canterbury","geom","TEXT",None,None,100,'',"NULLABLE","NON_REQUIRED",'').getOutput(0)
# Odvodit GeoJSON
with arcpy.da.UpdateCursor(Canterbury,['shape@','geom']) as cursor:
for row in cursor:
row[1] = str(row[0].__geo_interface__)
cursor.updateRow(row)
# Odstranit geometrii vytvořením tabulky
esriTable = arcpy.conversion.TableToTable(Canterbury,"memory","CanterburyTable").getOutput(0)
# Vytvořit arrow tabulku
arrowTable = arcpy.da.TableToArrowTable(esriTable)
# Zapsat parquet
pq.write_table(arrowTable,r'C:\Work\Parquet\TitlesCanterbury.parquet',
version='1.0',
compression='SNAPPY')<\/code><\/pre> <\/P>Bavte se s přesunem těchto dat ve velkém měřítku!<\/P> <\/P>
