Caching von Ergebnisfragmenten in Spark

Amazon EMR 6.6.0 und höher enthalten das optionale Feature zum Caching von Ergebnisfragmenten in Spark, mit dem Ergebnisfragmente automatisch zwischengespeichert werden. Diese Ergebnisfragmente sind Teile von Ergebnissen aus Teilbäumen von Abfragen, die in einem Amazon-S3-Bucket Ihrer Wahl gespeichert sind. Die gespeicherten Fragmente der Abfrageergebnisse werden bei nachfolgenden Abfrageausführungen wiederverwendet, was zu schnelleren Abfragen führt.

Beim Caching von Ergebnisfragmenten werden Ihre Spark-SQL-Abfragen analysiert und geeignete Ergebnisfragmente an Ihrem angegebenen S3-Speicherort zwischengespeichert. Bei nachfolgenden Abfrageläufen werden die verwendbaren Fragmente der Abfrageergebnisse automatisch erkannt und von S3 abgerufen. Das Caching von Ergebnisfragmenten unterscheidet sich vom Zwischenspeichern von Ergebnismengen, bei dem nachfolgende Abfragen exakt mit der ursprünglichen Abfrage übereinstimmen müssen, um Ergebnisse aus dem Cache zurückzugeben. Wenn es für Abfragen verwendet wird, die wiederholt auf eine statische Teilmenge Ihrer Daten abzielen, beschleunigt das Caching von Ergebnisfragmenten die Leistung erheblich.

Stellen Sie sich die folgende Abfrage vor, die Bestellungen bis zum Jahr 2022 zählt:

select
    l_returnflag,
    l_linestatus,
    count(*) as count_order
from
    lineitem
where
    l_shipdate <= current_date
    and year(l_shipdate) == '2022'
group by
    l_returnflag,
    l_linestatus

Im Laufe der Zeit muss diese Abfrage täglich ausgeführt werden, um den Gesamtumsatz des Jahres zu melden. Ohne das Caching von Ergebnisfragmenten müssen die Ergebnisse für alle Tage des Jahres täglich neu berechnet werden. Die Abfrage wird mit der Zeit langsamer und am Jahresende, wenn die Ergebnisse aller 365 Tage neu berechnet werden müssen, am langsamsten.

Wenn Sie das Caching von Ergebnisfragmenten aktivieren, verwenden Sie Ergebnisse für alle vorherigen Tage des Jahres aus dem Cache. An jedem Tag muss das Feature nur die Ergebnisse eines Tages neu berechnen. Nachdem das Feature das Ergebnisfragment berechnet hat, speichert das Feature das Fragment im Cache. Dadurch sind die Cache-aktivierten Abfragezeiten schnell und sie bleiben bei jeder nachfolgenden Abfrage konstant.

Caching von Ergebnisfragmenten in Spark aktivieren

Zum Aktivieren von Caching von Ergebnisfragmenten in Spark führen Sie die folgenden Schritte durch:

1. Erstellen Sie einen Cache-Bucket in Amazon S3 und autorisieren Sie den Lese-/Schreibzugriff für EMRFS. Weitere Informationen finden Sie unter Zugriff auf EMRFS-Daten in Amazon S3 genehmigen.

2. Stellen Sie die Amazon-EMR-Spark-Konfiguration ein, um das Feature zu aktivieren.

   spark.subResultCache.enabled = true
   spark.subResultCache.fs.root.path = s3://DOC-EXAMPLE-BUCKET/cache_dir/

3. Aktivieren Sie das S3-Lebenszyklusmanagement für den Bucket, um Cache-Dateien automatisch zu bereinigen.

4. Optional können Sie die maxBufferSize Eigenschaften reductionRationThreshold und konfigurieren, um die Funktion weiter zu optimieren.

   spark.sql.subResultCache.reductionRatioThreshold
   spark.sql.subResultCache.maxBufferSize

Einige Hinweise zur Verwendung von Caching von Ergebnisfragmenten

Die Kosteneinsparungen, wenn Sie Ergebnisse verwenden, die bereits in Amazon S3 zwischengespeichert wurden, anstatt sie neu zu berechnen, steigen mit der Häufigkeit, mit der dieselben zwischengespeicherten Ergebnisse verwendet werden können. Abfragen mit umfangreichen Tabellenscans, gefolgt von Filtern oder Hash-Aggregationen, die die Ergebnisgröße um den Faktor 8 reduzieren (d. h. ein Verhältnis von mindestens 8:1 bei Eingabegröße zu Ergebnissen), profitieren am meisten von dieses Features. Je größer das Reduktionsverhältnis zwischen Eingabe und Ergebnissen ist, desto größer ist der Kostenvorteil. Abfragen mit kleineren Reduktionsraten, die jedoch teure Rechenschritte zwischen dem Tabellenscan und Filtern oder Aggregationen enthalten, werden ebenfalls davon profitieren, sofern die Kosten für die Erstellung der Ergebnisse höher sind als die Kosten für das Abrufen der Ergebnisse aus Amazon S3. Standardmäßig wird das Caching von Ergebnisfragmenten nur wirksam, wenn erkannt wird, dass das Reduktionsverhältnis mindestens 8:1 beträgt.

Wenn Ihre Abfragen zwischengespeicherte Ergebnisse wiederholt wiederverwenden, sind die Vorteile dieses Features am größten. Rollende und inkrementelle Fensterabfragen sind gute Beispiele. Beispielsweise müsste bei einer 30-tägigen rollenden Fenster-Abfrage, die bereits 29 Tage lang ausgeführt wurde, nur 1/30 der Zieldaten aus der ursprünglichen Eingabequelle abgerufen werden, und es würden zwischengespeicherte Ergebnisfragmente für die 29 vorherigen Tage verwendet. Eine inkrementelle Fensterabfrage wäre sogar noch besser, da der Beginn des Fensters unverändert bleibt: Bei jedem Aufruf der Abfrage muss ein kleinerer Prozentsatz der Verarbeitung aus der Eingabequelle gelesen werden.

Im Folgenden finden Sie weitere Überlegungen zur Verwendung von Caching von Ergebnisfragmenten:

• Abfragen, die nicht auf dieselben Daten mit denselben Abfragefragmenten abzielen, haben eine niedrige Cache-Trefferrate und profitieren daher nicht von diesem Feature.

• Abfragen mit niedrigen Reduktionsraten, die keine teuren Rechenschritte enthalten, führen zu zwischengespeicherten Ergebnissen, deren Lesen ungefähr so teuer ist wie die ursprüngliche Verarbeitung.

• Die erste Abfrage weist aufgrund der Kosten für das Schreiben in den Cache immer eine geringfügige Regression auf.

• Das Feature zum Caching von Ergebnisfragmenten funktioniert ausschließlich mit Parquet-Dateien. Andere Dateiformate werden nicht unterstützt.

• Das Feature Caching von Ergebnisfragmenten für das Zwischenspeichern von Ergebnisfragmenten versucht nur, Scans mit Datei-Split-Größen von 128 MB oder mehr zwischenzuspeichern. Bei der Standardkonfiguration von Spark ist das Caching von Ergebnisfragmenten deaktiviert, wenn die Scangröße (Gesamtgröße aller gescannten Dateien) geteilt durch die Anzahl der Executor-Cores weniger als 128 MB beträgt. Wenn eine der unten aufgeführten Spark-Konfigurationen festgelegt ist, beträgt die Größe der aufgeteilten Datei:

  min(maxPartitionBytes, max(openCostInBytes, scan size / minPartitionNum))

  • spark.sql. leafNodeDefaultParallelismus (Standardwert ist spark.default.parallelism)

  • spark.sql.files. minPartitionNum (Der Standardwert ist spark.sql. leafNodeDefaultParallelität)

  • spark.sql.files. openCostInByte

  • spark.sql.dateien. maxPartitionBytes

• Das Feature zum Caching von Ergebnisfragmenten speichert die Granularität der RDD-Partition im Cache. Das zuvor beschriebene Reduktionsverhältnis, das standardmäßig 8:1 beträgt, wird pro RDD-Partition bewertet. Bei Workloads mit Reduktionsraten pro RDD, die sowohl größer als auch kleiner als 8:1 sind, kann es zu geringeren Leistungsvorteilen kommen als bei Workloads mit Reduktionsraten pro RDD, die durchweg unter 8:1 liegen.

• Das Feature zum Caching von Ergebnisfragmenten verwendet standardmäßig einen Schreibpuffer von 16 MB für jede RDD-Partition, die zwischengespeichert wird. Wenn mehr als 16 MB pro RDD-Partition zwischengespeichert werden, können die Kosten für die Feststellung, dass kein Schreibvorgang möglich ist, zu Leistungsregressionen führen.

• Standardmäßig versucht Caching von Ergebnisfragmenten zwar nicht, Ergebnisse einer RDD-Partition mit einem Reduktionsverhältnis von weniger als 8:1 zwischenzuspeichern und seinen Schreibpuffer auf 16 MB begrenzt, aber beide Werte können über die folgenden Konfigurationen eingestellt werden:

  spark.sql.subResultCache.reductionRatioThreshold (default: 8.0)
  spark.sql.subResultCache.maxBufferSize (default: 16MB, max: 64MB)

• Mehrere Cluster, die dieselbe Amazon-EMR-Version verwenden, können sich denselben Cache-Speicherort teilen. Um die Richtigkeit der Ergebnisse sicherzustellen, verwendet Caching von Ergebnisfragmenten keine Cache-Ergebnisse, die von verschiedenen Versionen von Amazon EMR geschrieben wurden.

• Das Zwischenspeichern von Ergebnisfragmenten wird automatisch für Spark Streaming-Anwendungsfälle deaktiviert oder wenn RecordServer Apache Ranger oder verwendet AWS Lake Formation wird.

• Beim Lesen/Schreiben des Ergebnisfragment-Cache werden EMRFS- und Amazon-S3-Buckets verwendet. CSE/ SSE S3/SSE KMS-Verschlüsselung wird unterstützt.