Ein Kollege fragte mich kürzlich, wie man ein paar Milliarden Datensätze in GeoEvents spatiotemporalen Big Data Store (STBDS) bei einem Kunden vor Ort verschieben kann, unter Verwendung von ArcGIS Data Interoperability. Dabei handelt es sich um archivierte Fahrzeugpositionen für ein Versorgungsunternehmen, die in einer Oracle-Geodatenbank gespeichert sind. Täglich kommen ein paar Millionen neue Ereignisse hinzu. Die archivierten Daten befinden sich in über 70 Tabellen.<\/P>
Da ein Feature-Service im STBDS dieselbe REST-API wie ein gewöhnlicher gehosteter Feature-Service hat, war der Plan von Anfang an, das im meinem früheren Blog beschriebene Muster zu nutzen, nämlich die Archivdaten mittels ETL in Portal-Shapefile-Elemente zu laden und dann den Append-REST-Endpunkt der Ziel-Layer asynchron zu verwenden, um die Daten zu laden. Eine zusätzliche Herausforderung bestand darin, das ETL parallel als mehrere gleichzeitige Jobs auszuführen. Dieser Ansatz würde das Risiko von Netzwerkausfällen beim Streaming kleiner Transaktionen (1000 Features ist die Standard-Batch-Größe) zum Portal reduzieren. Es stellte sich heraus, dass die Kundenumgebung nicht auf einer Version war, die den Workflow unterstützte, und wir haben tatsächlich Streaming verwendet, aber es brachte mich dazu, darüber nachzudenken, wie man Big Data in die neue Welle von Cloud-Warehouses ETL-en kann, die Sie nativ und in ArcGIS Pro 2.9 verwenden können.<\/P>
Meine Grundbotschaft ist, dass Sie diese Lift-and-Shift-Jobs durchführen können, indem Sie große Datensätze als Dateien in mehreren gleichzeitigen Jobs verschieben und so den Durchsatz maximieren und das Transport-Risiko minimieren. Sehen wir uns an wie.<\/STRONG><\/P>Cloud-Warehouses wie Snowflake, Big Query und Redshift können in ArcGIS Pro 2.9 abgefragt und gelesen werden, aber nicht mit Standardwerkzeugen beschrieben werden. ArcGIS Data Interoperability kann jedoch in diese Warehouses schreiben, einschließlich räumlicher Daten, aber der Standardmodus ist Streaming, was möglicherweise nicht so skaliert, wie Sie es benötigen. Ich werde Ihnen ein Muster zeigen, das Sie für alle drei Warehouses verwenden können:<\/P>ETL ganzer Datensätze, einschließlich räumlicher Daten, als Apache Parquet oder anderes Format wie CSVGeometrie wird in einem zeichenbasierten Standardformat kodiert<\/LI><\/UL><\/LI>Automatisieren Sie das ETL in mehreren gleichzeitigen Prozessen<\/STRONG><\/LI>Schreiben Sie keinen Code, außer dem erforderlichen SQL oder Makrobefehlen für die Zielumgebung<\/LI><\/UL>Ich werde auch allen Programmierern in meinem No-Code-Blog-Bereich einen kleinen Tipp geben, siehe unten 😉<\/span>, also einige Python-Tipps zur Erstellung von Parquet-Dateien (Hinweis: Parquet-Dateien sind seit dem 22. September 2021 ein unterstützter Elementtyp in ArcGIS Online. Teilen Sie welche!)
<\/span} Milliarden von Features sind mit diesem Muster möglich, aber ich verwende für Demonstrationszwecke einen bescheideneren Datensatz mit nur 2,3 Millionen Punktfeatures.<\/P>
2.3 Millionen Punktfeatures<\/span><\/span><\/P>Meine Daten befinden sich in 12 Feature-Klassen, Sie können jede beliebige Anzahl haben. Das Muster, das ich zeigen werde, funktioniert mit Daten, die in separate Teile aufgeteilt sind und gleichzeitig verarbeitet werden können. Wenn Ihre Daten monolithisch sind, teilen Sie sie entweder selbst räumlich (orientiertes Fischernetz anyone?<\/A>) oder durch Hinzufügen eines Feldes mit einer Batch-Kennung basierend auf der Zeilenposition – dieses Feld können Sie während der Verarbeitung entfernen.<\/P>Ich erwähnte Snowflake-, Big Query- und Redshift-Warehouses. In allen Fällen können Sie Parquet-Dateien dort ablegen, wo die Zielumgebung sie sehen kann und dann aus den Parquet-Dateien laden. Für räumliche Daten müssen die Parquet-Dateien Geometrie in einem vom Zielsystem verstandenen Format kodieren (
Snowflake<\/A> und Big Query<\/A> unterstützen GeoJSON und WKT, Redshift<\/A> unterstützt WKT). Ich werde nur ein Beispiel mit GeoJSON zeigen, das zu Snowflake geladen wird. Meine Demo-Daten sind Punktgeometrie und das Feld zur Speicherung des GeoJSON hat eine Breite von 100. Wenn Sie Linien- oder Polygon-Daten verwenden, sollten Sie untersuchen, wie breit Ihre punktreichsten Features beim Kodieren sind. Zum Beispiel habe ich ein sehr punktreiches Polygon in einem Layer ausgewählt und als GeoJSON hat es 2.071.156 Zeichen:<\/P> <\/P>
with arcpy.da.SearchCursor('NZ Property Titles','shape@') as cursor:
for row in cursor:
print(len(str(row[0].__geo_interface__)))
2071156<\/code><\/pre><P> <\/P><P>Beachten Sie, dass Data Interoperability die Dezimalgenauigkeit steuern kann, die von GeoJSON verwendet wird; für geografische Daten ist ein Wert von 7 angemessen. Dasselbe Polygon verwendet dann <SPAN>1.274.064 Zeichen. Zum Beispiel geht die erste Koordinate von <SPAN>(172.90677540000001,-41.12752416699993) <\SPAN>zu (<SPAN>172.9067754,-41.1275242). Denken Sie daran: Jedes Byte zählt!<\SPAN></SPAN></P><P><STRONG>Hinweis:<\STRONG> Für Big Query verfügt Data Interoperability über einen <A title="GoogleBigQueryConnector" href="https:\/\/docs.safe.com\/fme\/html\/FME_Desktop_Documentation\/FME_Transformers\/Transformers\/googlebigqueryconnector-pkg.htm" target="_blank" rel="noopener nofollow noreferrer">GoogleBigQueryConnector<\A>, der CSV-Dateien in Tabellen laden kann. Dies könnte einfacher sein als Parquet zu senden und die bq-Befehlsumgebung zum Laden der Daten zu verwenden. Ich habe dieses Szenario nicht untersucht.<\P><P>Lassen Sie uns meinen speziellen Workflow genauer betrachten. Das Geheimnis ist es, <STRONG>zwei<\STRONG> ETL-Werkzeuge zu erstellen: Das erste verwaltet die Jobs und ruft einen <A title="WorkspaceRunner" href="https:\/\/docs.safe.com\/fme\/html\/FME_Desktop_Documentation\/FME_Transformers\/Transformers\/workspacerunner.htm" target="_blank" rel="noopener nofollow noreferrer">WorkspaceRunner<\A>-Transformer auf, der das zweite Werkzeug aufruft, welches die Arbeit erledigt. Es ist sehr einfach. Hier ist <STRONG>LoadManager.fmw<\STRONG>, es nimmt eine Liste von Argumenten entgegen – in meinem Fall Feature-Class-Namen in einer Geodatabase:<\P><P><span class="lia-inline-image-display-wrapper lia-image-align-center" image-alt="LoadManager" style="width: 400px;"><img src="https:\/\/us.v-cdn.net\/6038851\/uploads\/images\/23278i26931F24B6DB0008\/LoadManager.jpg" role="button" title="LoadManager.jpg" alt="LoadManager" \/><span class="lia-inline-image-caption" onclick="event.preventDefault();">LoadManager<\span><\span></P><P> WorkspaceRunner startet bis zu 7 FME-Prozesse, die ein Zielwerkzeug ausführen bis die Job-Warteschlange leer ist. Die Verarbeitung wird wahrscheinlich CPU-begrenzt sein während die Worker-Prozesse Datensatzdateien extrahieren, kodieren und hochladen. Ich ließ jeden Prozess zwei Jobs ausführen, was meine eingehenden Datensätze mit 6 Prozessen verarbeitete.<\P><P><span class="lia-inline-image-display-wrapper lia-image-align-center" image-alt="WorkspaceRunner" style="width: 400px;"><img src="https:\/\/us.v-cdn.net\/6038851\/uploads\/images\/23279i73187AEDC185DFB7\\WorkspaceRunner.jpg" role=\button\ title=\WorkspaceRunner.jpg\ alt=\WorkspaceRunner\ \/><span class=\lia-inline-image-caption\ onclick=\event.preventDefault();\>WorkspaceRunner<\span><\span></P><P> <\P><P>Hier ist <STRONG>LoadWorkerParquet.fmw<\STRONG>.<\P><P><span class=\lia-inline-image-display-wrapper lia-image-align-center\ image-alt=\LoadWorkerParquet\ style=\width: 400px;\"><img src=\https:\/\\us.v-cdn.net\\6038851\\uploads\\images\\23280i68720B57C8DAA789\\LoadWorkerParquet.jpg\ role=button title=LoadWorkerParquet.jpg alt=LoadWorkerParquet \/><span class=lia-inline-image-caption onclick=event.preventDefault();>LoadWorkerParquet<\span></span></p><p>Es ist ebenfalls ein einfaches Werkzeug: Es liest eine Geodatabase aus, schreibt eine lokale Parquet-Datei und sendet diese dann an Snowflake, wo die Daten in eine Tabelle kopiert werden. Ich lasse Sie den SQLExecutor selbst inspizieren aber zur besseren Verständlichkeit hier nach Variablenersetzung ein Beispiel für eine Anweisung (für Snowflake):<br /><pre class=lia-code-sample language-sql"><code>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);
Sobald die Parquet-Datei bei Snowflake angekommen ist, erfolgt das Einlesen blitzschnell. Übrigens habe ich gelernt dies zu tun, indem Sie die Hilfe lesen, ich bin kein Snowflake DBA.
Welche Leistung sollten Sie erwarten? Zum Zeitpunkt des Schreibens sitze ich wie der Rest von uns zu Hause fest, aber mit meinem Heim-WLAN lade ich 2,3 Millionen Features in 6 Minuten in Snowflake, also denke ich konservativ an 25 Millionen point-Features pro Stunde mit einem anständigen Computer und kabelgebundenem Netzwerk. Natürlich könnten Sie für eine Produktionsumgebung und einen wirklich großen Job mehrere Computer verwenden, sicherlich skalieren die Ziel-Cloud-Warehouses, um den Durchsatz zu bewältigen.
Im Blog-Download werden Sie feststellen, dass ich ein zweites Worker-Tool, LoadWorker.fmw, beifüge. Dies diente mir dazu, die Leistung der üblichen Methode zum Schreiben in Snowflake mit 100K Features pro Transaktion zu vergleichen; es war viel langsamer.
Jetzt, zurück in core Pro 2.9, sind meine Daten in Snowflake geladen und ich kann Abfragen darauf ausführen und das skalierte Compute-Erlebnis genießen.
Snowflake im Katalogbereich
Ich erwähnte eine Python-Option zum Erstellen von parquet-Dateien, sie ist im Blog-Download enthalten, aber hier ist sie auch:
# Pro 2.9+ Beispiel zur Erstellung einer parquet-Datei aus einer Feature-Class
# Geometrie wird als GeoJSON in einem Feld 'geom' codiert
import arcpy
import pyarrow.parquet as pq
arcpy.env.overwriteOutput = True
# Quell-Feature-Class
Canterbury = r"C:\Work\Parquet\Parquet.gdb\Canterbury"
# Erstelle In-Memory Feature-Class in 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")
# Füge das geom-Feld hinzu (Nicht-Punkt-Geometrie erfordert ein breiteres Feld)
Canterbury = arcpy.management.AddField("memory\Canterbury","geom","TEXT",None,None,100,'',"NULLABLE","NON_REQUIRED",'').getOutput(0)
# Leite GeoJSON ab
with arcpy.da.UpdateCursor(Canterbury,['shape@','geom']) as cursor:
for row in cursor:
row[1] = str(row[0].__geo_interface__)
cursor.updateRow(row)
# Entferne Geometrie durch Erstellen einer Tabelle
esriTable = arcpy.conversion.TableToTable(Canterbury,"memory","CanterburyTable").getOutput(0)
# Erstelle Arrow-Tabelle
arrowTable = arcpy.da.TableToArrowTable(esriTable)
# Schreibe parquet
pq.write_table(arrowTable,r'C:\Work\Parquet\TitlesCanterbury.parquet',
version='1.0',
compression='SNAPPY')
Viel Spaß beim Verschieben dieser Daten im großen Maßstab!
