Einführung und Architektur von Apache Arrow, einem leistungsstarken Datenformatbibliothekspaket auf JVM (Gkatziouras)

Apache Arrow ist ein beliebtes Format, das von verschiedenen Big Data-Tools, einschließlich BigQuery, verwendet wird und ein Speicherformat sowohl für flache als auch hierarchische Daten ist. Es handelt sich um eine speicherintensive Methode zum Beschleunigen von Anwendungen.

Eine häufig verwendete Bibliothek im Bereich der Datenverarbeitung und Datenwissenschaft: Apache Arrow. Arrow wird von Open-Source-Projekten wie Apache Parquet, Apache Spark, Pandas und vielen kommerziellen oder Closed-Source-Diensten verwendet. Es bietet die folgenden Funktionen:

• In-Memory-Computing
• Standardisiertes spaltenbasiertes Speicherformat
• Ein IPC- und RPC-Framework für den Datenaustausch zwischen Prozessen bzw. Knoten
  

Werfen wir einen Blick darauf, wie die Dinge vor der Einführung von Arrow funktionierten:

Wir können sehen, dass wir die Daten im Parquet-Format lesen und deserialisieren müssen, damit Spark Daten aus einer Parquet-Datei lesen kann. Dies erfordert, dass wir eine vollständige Kopie der Daten erstellen, indem wir sie in den Speicher laden. Zuerst lesen wir die Daten in einen In-Memory-Puffer und verwenden dann die Konvertierungsmethoden von Parquet, um die Daten (z. B. einen String oder eine Zahl) in eine Darstellung unserer Programmiersprache umzuwandeln. Dies ist notwendig, da Parquet Zahlen anders darstellt als die Programmiersprache Python.

Dies stellt aus mehreren Gründen ein großes Leistungsproblem dar:

• Wir kopieren die Daten und führen die Transformationsschritte darauf aus. Die Daten liegen in unterschiedlichen Formaten vor und wir müssen alle Daten lesen und konvertieren, bevor wir Berechnungen damit durchführen können.
• Die Daten, die wir laden, müssen in den Speicher passen. Du hast nur 8GB RAM und die Daten sind 10GB groß? Du hast so ein Pech!
  

Sehen wir uns nun an, wie Apache Arrow dies verbessert:

Anstatt Daten zu kopieren und zu transformieren, versteht Arrow es, Daten direkt zu lesen und zu bearbeiten. Zu diesem Zweck hat die Arrow-Community ein neues Dateiformat und neue Operationen definiert, die direkt auf den serialisierten Daten arbeiten. Dieses Datenformat kann direkt von der Festplatte gelesen werden, ohne dass es in den Speicher geladen und die Daten konvertiert/deserialisiert werden müssen. Natürlich wird ein Teil der Daten trotzdem in den RAM geladen, aber Ihre Daten müssen nicht in den Speicher passen. Arrow nutzt seine Dateispeicherzuordnungsfunktionen, um nur so viele Daten wie nötig und möglich in den Speicher zu laden.

Apache Arrow unterstützt die folgenden Sprachen:

• C++
• C#
• Gehen
• Java
• JavaScript
• Rost
• Python (über die C++-Bibliothek)
• Ruby (über die C++-Bibliothek)
• R (über die C++-Bibliothek)
• MATLAB (über die C++-Bibliothek).
  

Arrow-Funktionen

Arrow ist in erster Linie eine Bibliothek, die spaltenbasierte Datenstrukturen für In-Memory-Computing bereitstellt. Alle Daten können dekomprimiert und in Arrow-spaltenbasierte Datenstrukturen dekodiert werden, sodass die dekodierten Daten anschließend einer In-Memory-Analyse unterzogen werden können. Das Arrow-Spaltenformat weist einige nette Eigenschaften auf: Der wahlfreie Zugriff ist O(1) und jede Wertezelle grenzt im Speicher an die vorherige und nächste, sodass die Iteration sehr effizient ist.

Apache Arrow definiert ein binäres „Serialisierungs“-Protokoll zum Anordnen von Sammlungen von Arrow-Spalten-Arrays (sogenannte „Datensatz-Batches“), die für Messaging und Interprozesskommunikation verwendet werden können. Sie können das Protokoll überall ablegen, auch auf der Festplatte, und es später im Speicher abbilden oder in den Speicher einlesen und an einen anderen Ort senden.

Das Arrow-Protokoll ist so konzipiert, dass Sie einen Block von Arrow-Daten ohne Deserialisierung „abbilden“ können. Bei der Analyse von Arrow-Protokolldaten auf der Festplatte können Sie daher die Speicherzuordnung verwenden und es fallen praktisch keine Kosten an. Dieses Protokoll wird für viele Dinge verwendet, beispielsweise zum Streamen von Daten zwischen Spark SQL und Python oder zum Ausführen von Pandas-Funktionen für Spark SQL-Datenblöcke. Diese werden als „Pandas-UDFs“ bezeichnet.

Arrow ist für den Speicher konzipiert (Sie können es jedoch auf die Festplatte legen und dann dem Speicher zuordnen). Sie sind so konzipiert, dass sie untereinander kompatibel sind und gemeinsam in Anwendungen verwendet werden können, während ihre Konkurrenten, die Apache Parquet-Dateien, für die Festplattenspeicherung konzipiert sind.

Vorteile: Apache Arrow definiert ein sprachunabhängiges spaltenorientiertes Speicherformat für flache und hierarchische Daten, das für effiziente Analysevorgänge auf moderner Hardware wie CPUs und GPUs organisiert ist. Das Arrow-Speicherformat unterstützt außerdem Zero-Copy-Lesevorgänge für blitzschnellen Datenzugriff ohne Serialisierungs-Overhead.

Apache Arrow für Java

Importieren Sie die Bibliothek:

<Abhängigkeit>
    <groupId>org.apache.arrow</groupId>
    <artifactId>Pfeil-Speicher-Netty</artifactId>
    <version>${arrow.version}</version>
</Abhängigkeit>
<Abhängigkeit>
    <groupId>org.apache.arrow</groupId>
    <artifactId>Pfeil-Vektor</artifactId>
    <version>${arrow.version}</version>
</Abhängigkeit>

Bevor wir beginnen, ist es wichtig zu verstehen, dass für Arrow-Lese-/Schreibvorgänge Byte-Puffer verwendet werden. Bei Vorgängen wie Lesen und Schreiben handelt es sich um einen kontinuierlichen Byte-Austausch. Zur Verbesserung der Effizienz verfügt Arrow über einen Pufferallokator, der eine feste Größe haben oder automatisch erweitert werden kann. Bibliotheken, die die Zuweisungsverwaltung unterstützen, sind Arrow-Memory-Netty und Arrow-Memory-Unsafe. Wir verwenden hier Netty.

Zum Speichern von Daten mit Arrow ist ein Schema erforderlich, das programmgesteuert definiert werden kann:

Paket com.gkatzioura.arrow;

importiere java.io.IOException;

importiere java.util.List;

importiere org.apache.arrow.vector.types.pojo.ArrowType;

importiere org.apache.arrow.vector.types.pojo.Field;

importiere org.apache.arrow.vector.types.pojo.FieldType;

importiere org.apache.arrow.vector.types.pojo.Schema;

öffentliche Klasse SchemaFactory {

öffentliches statisches Schema DEFAULT_SCHEMA = createDefault();

öffentliches statisches Schema createDefault() {

var strField = neues Feld("col1", FieldType.nullable(neuer ArrowType.Utf8()), null);

var intField = neues Feld("col2", FieldType.nullable(neuer ArrowType.Int(32, true)), null);

gib ein neues Schema zurück (Liste von (strField, intField));

}

öffentliches statisches Schema schemaWithChildren() {

var Betrag = neues Feld("Betrag", FieldType.nullable(neuer ArrowType.Decimal(19,4,128)), null);

var Währung = neues Feld("Währung",FieldType.nullable(neuer ArrowType.Utf8()), null);

var itemField = neues Feld("item", FieldType.nullable(neuer ArrowType.Utf8()), List.of(Betrag, Währung));

gib ein neues Schema zurück (Liste von (Elementfeld));

}

öffentliches statisches SchemafromJson(String jsonString) {

versuchen {

return Schema.fromJSON(jsonString);

} Fang (IOException e) {

wirf eine neue ArrowExampleException(e);

}

}

}

Sie verfügen auch über eine analysierbare JSON-Darstellung:

{
  "Felder" : [ {
    "Name": "Spalte1",
    "nullable" : wahr,
    "Typ" : {
      "Name": "utf8"
    },
    "Kinder" : [ ]
  }, {
    "Name": "Spalte2",
    "nullable" : wahr,
    "Typ" : {
      "Name" : "int",
      "Bitbreite" : 32,
      "istSigniert" : wahr
    },
    "Kinder" : [ ]
  } ]
}

Darüber hinaus können Sie genau wie Avro komplexe Schemata und eingebettete Werte in Feldern entwerfen:

öffentliches statisches Schema schemaWithChildren() {
    var Betrag = neues Feld("Betrag", FieldType.nullable(neuer ArrowType.Decimal(19,4,128)), null);
    var Währung = neues Feld("Währung",FieldType.nullable(neuer ArrowType.Utf8()), null);
    var itemField = neues Feld("item", FieldType.nullable(neuer ArrowType.Utf8()), List.of(Betrag, Währung));
 
    gib ein neues Schema zurück (Liste von (Elementfeld));
}

Basierend auf dem obigen Schema erstellen wir ein DTO für unsere Klasse:

Paket com.gkatzioura.arrow;
 
lombok.Builder importieren;
importiere lombok.Data;
 
@Daten
@Erbauer
öffentliche Klasse DefaultArrowEntry {
 
    private Zeichenfolge col1;
    private Integer col2;
 
}

Unser Ziel ist es, diese Java-Objekte in Arrow-Byte-Streams zu konvertieren.

1. Erstellen Sie einen DirectByteBuffer mit einem Allocator

Diese Puffer sind Off-Heap. Sie müssen den verwendeten Speicher freigeben, für den Bibliotheksbenutzer geschieht dies jedoch durch Ausführen einer close()-Operation am Allocator. In unserem Fall implementiert unsere Klasse die Schnittstelle Closeable, die den Allocator-Schließvorgang durchführt.

Durch Verwendung der Streaming-API werden die Daten an einen OutPutStream gestreamt, der im Arrow-Format übermittelt wird:

Paket com.gkatzioura.arrow;
 
importiere java.io.Closeable;
importiere java.io.IOException;
importiere java.nio.channels.WritableByteChannel;
importiere java.util.List;
 
importiere org.apache.arrow.memory.RootAllocator;
importiere org.apache.arrow.vector.IntVector;
importiere org.apache.arrow.vector.VarCharVector;
importiere org.apache.arrow.vector.VectorSchemaRoot;
importiere org.apache.arrow.vector.dictionary.DictionaryProvider;
importiere org.apache.arrow.vector.ipc.ArrowStreamWriter;
importiere org.apache.arrow.vector.util.Text;
 
importiere statisches com.gkatzioura.arrow.SchemaFactory.DEFAULT_SCHEMA;
 
öffentliche Klasse DefaultEntriesWriter implementiert Closeable {
 
    privater endgültiger RootAllocator rootAllocator;
    private final VectorSchemaRoot vectorSchemaRoot; //Erstellung des Vektor-Allocators:
 
    öffentliche DefaultEntriesWriter() {
        rootAllocator = neuer RootAllocator();
        vectorSchemaRoot = VectorSchemaRoot.create(DEFAULT_SCHEMA, rootAllocator);
    }
 
    public void write(List<DefaultArrowEntry> defaultArrowEntries, int batchSize, WritableByteChannel out) {
        wenn (Batchgröße <= 0) {
            batchSize = defaultArrowEntries.size();
        }
 
        DictionaryProvider.MapDictionaryProvider dictProvider = neuer DictionaryProvider.MapDictionaryProvider();
        versuche(ArrowStreamWriter writer = neuer ArrowStreamWriter(vectorSchemaRoot, dictProvider, out)) {
            Autor.start();
 
            VarCharVector childVector1 = (VarCharVector) vectorSchemaRoot.getVector(0);
            IntVector childVector2 = (IntVector) vectorSchemaRoot.getVector(1);
            childVector1.reset();
            childVector2.reset();
 
            boolean exactBatches = defaultArrowEntries.size()%batchSize == 0;
            Int Batchzähler = 0;
 
            für(int i=0; i < defaultArrowEntries.size(); i++) {
                childVector1.setSafe(batchCounter, neuer Text(defaultArrowEntries.get(i).getCol1()));
                childVector2.setSafe(batchCounter, defaultArrowEntries.get(i).getCol2());
 
                batchZähler++;
 
                wenn(batchZähler == batchGröße) {
                    vectorSchemaRoot.setRowCount(batchSize);
                    Schriftsteller.writeBatch();
                    Stapelzähler = 0;
                }
            }
 
            wenn(!exactBatches) {
                } vectorSchemaRoot.setRowCount(batchCounter);
                Schriftsteller.writeBatch();
            }
 
            Schriftsteller.Ende();
        } Fang (IOException e) {
            wirf eine neue ArrowExampleException(e);
        }
    }
 
    @Überschreiben
    public void close() wirft IOException {
        VektorSchemaRoot.close();
        rootAllocator.schließen();
    }
 
}

Um die Batch-Unterstützung auf Arrow anzuzeigen, wurde in der Funktion ein einfacher Batch-Algorithmus implementiert. Betrachten wir für unser Beispiel einfach das Schreiben der Daten in Stapeln.

Schauen wir uns genauer an, was der obige Code macht:

Erstellen eines Vektor-Allocators:

öffentlicher DefaultEntriesToBytesConverter() {
    rootAllocator = neuer RootAllocator();
    vectorSchemaRoot = VectorSchemaRoot.create(DEFAULT_SCHEMA, rootAllocator);
}

Beim Schreiben in den Stream wird dann ein Arrow Stream Writer implementiert und gestartet

ArrowStreamWriter-Writer = neuer ArrowStreamWriter (vectorSchemaRoot, dictProvider, Channels.newChannel (out));
Autor.start();

Wir füllen die Vektoren mit Daten und setzen sie dann ebenfalls zurück, lassen aber die vorab zugewiesenen Puffer an Ort und Stelle:

VarCharVector childVector1 = (VarCharVector) vectorSchemaRoot.getVector(0);
IntVector childVector2 = (IntVector) vectorSchemaRoot.getVector(1);
childVector1.reset();
childVector2.reset();

Beim Schreiben von Daten verwenden wir die Operation setSafe. Dies sollte getan werden, wenn mehr Puffer zugewiesen werden müssen. In diesem Beispiel wird dies bei jedem Schreibvorgang durchgeführt, kann aber vermieden werden, wenn die erforderlichen Vorgänge und Puffergrößen berücksichtigt werden:

childVector1.setSafe(i, neuer Text(defaultArrowEntries.get(i).getCol1()));
childVector2.setSafe(i, defaultArrowEntries.get(i).getCol2());

Schreiben Sie dann den Batch in den Stream:

vectorSchemaRoot.setRowCount(batchSize);
Schriftsteller.writeBatch();

Zu guter Letzt schließen wir den Autor:

@Überschreiben
public void close() wirft IOException {
    VektorSchemaRoot.close();
    rootAllocator.schließen();
}

Oben finden Sie ausführliche Informationen zur Einführung und Architektur von Apache Arrow, einem leistungsstarken Datenformatbibliothekspaket auf JVM (Gkatziouras). Weitere Informationen zum Einstieg in Apache Arrow finden Sie in den anderen verwandten Artikeln auf 123WORDPRESS.COM!