ez a blog elmagyarázza, hogyan kell írni egy DataFrame egyetlen fájlba Spark. Azt is leírja, hogyan lehet adatokat írni egy adott névvel rendelkező fájlba, ami meglepően kihívást jelent.
egyetlen fájl Spark-val történő kiírása nem jellemző. A Spark célja több fájl párhuzamos kiírása. Sok fájl egyidejű kiírása gyorsabb a nagy adatkészleteknél.
alapértelmezett viselkedés
hozzunk létre egy Adatkeretet, a repartition(3) használatával hozzunk létre három memóriapartíciót, majd írjuk ki a fájlt a lemezre.
val df = Seq("one", "two", "three").toDF("num")df .repartition(3) .write.csv(sys.env("HOME")+ "/Documents/tmp/some-files")
itt vannak a lemezen generált fájlok.
Documents/ tmp/ some-files/ _SUCCESS part-00000-b69460e8-fdc3-4593-bab4-bd15fa0dad98-c000.csv part-00001-b69460e8-fdc3-4593-bab4-bd15fa0dad98-c000.csv part-00002-b69460e8-fdc3-4593-bab4-bd15fa0dad98-c000.csv
A Spark memória partíciónként egy fájlt ír ki. A repartition(3) – et három memóriapartíció létrehozásához használtuk, így három fájlt írtunk.
egy fájl kiírása
felosztással használhatjuk a repartition(1) egyetlen fájlt.
df .repartition(1) .write.csv(sys.env("HOME")+ "/Documents/tmp/one-file-repartition")
itt van a lemezre írt fájl.
Documents/ tmp/ one-file-repartition/ _SUCCESS part-00000-d5a15f40-e787-4fd2-b8eb-c810d973b6fe-c000.csv
nem tudjuk ellenőrizni az írott fájl nevét. Szabályozhatjuk a könyvtár nevét, de magát a fájlt nem.
ez a megoldás nem elegendő, ha egy adott nevű fájlba szeretne adatokat írni.
egyetlen fájl kiírása a coalesce
segítségével a coalesce(1) fájlt is írhatjuk.
df .coalesce(1) .write.csv(sys.env("HOME")+ "/Documents/tmp/one-file-coalesce")
itt van, mi outputted.
Documents/ tmp/ one-file-coalesce/ _SUCCESS part-00000-c7521799-e6d8-498d-b857-2aba7f56533a-c000.csv
a coalesce sem engedi, hogy egy adott fájlnevet állítsunk be (csak a mappa nevét szabjuk testre). A spark-daria-t kell használnunk egy olyan módszer eléréséhez, amely egyetlen fájlt ad ki.
meghatározott nevű fájl kiírása
A spark-daria-ban meghatározott DariaWriters.writeSingleFile függvény segítségével egyetlen fájlt írhat ki egy adott fájlnévvel.
itt van a kód, amely kiírja az adatkeret tartalmát a ~/Documents/better/mydata.csv fájlba.
import com.github.mrpowers.spark.daria.sql.DariaWritersDariaWriters.writeSingleFile( df = df, format = "csv", sc = spark.sparkContext, tmpFolder = sys.env("HOME") + "/Documents/better/tmp", filename = sys.env("HOME") + "/Documents/better/mydata.csv")
a writeSingleFile metódus nevezze el a fájlt anélkül, hogy aggódna a bonyolult megvalósítási részletek miatt.
writeSingleFile is használ repartition(1) és Hadoop fájlrendszer módszerek alatt a motorháztető. Az összes Hadoop fájlrendszer módszerek állnak rendelkezésre minden Spark runtime környezetben – nem kell csatolni külön Üvegek.
kompatibilitás más fájlrendszerekkel
a legjobb, ha a Hadoop fájlrendszer módszereket használja fájlok áthelyezésekor, átnevezésekor vagy törlésekor, így a kód több platformon fog működni. writeSingleFile működik a helyi fájlrendszeren és az S3-ban. Ezt a megközelítést akkor használhatja, ha a Sparkot helyben vagy Databricks jegyzetfüzetben futtatja.
vannak más megoldások erre a problémára, amelyek nem cross platform. Vannak olyan megoldások, amelyek csak a Databricks notebookokban működnek, vagy csak az S3-ban működnek, vagy csak Unix-szerű operációs rendszeren működnek.
a Hadoop fájlrendszer módszerek ügyetlen dolgozni, de a legjobb megoldás, mert a munka több platformon.
a writeSingleFile módszer a fs.rename() Hadoop módszert használja, a válaszban leírtak szerint. Itt van a psuedocode:
val src = new Path("s3a://bucket/data/src")val dest = new Path("s3a://bucket/data/dest")val conf = sc.hadoopConfiguration // assuming sc = spark contextval fs = src.getFileSystem(conf)fs.rename(src, dest)
a copyMerge
Hadoop 2 rendelkezik egy FileUtil.copyMerge() módszerrel, amely elegáns megoldás erre a problémára, de ez a módszer elavult, és eltávolításra kerül a Hadoop 3-ban. Ebben a szálban van egy válasz, amely copyMerge – et újraimplementál a Hadoop 3 felhasználók számára.
mindenesetre ne írjon olyan kódot, amely a FileUtil.copyMerge() módszerre támaszkodik. Tudjuk, hogy a módszer elérhetetlen lesz, amikor a Spark frissíti a Hadoop 3-at, és nem akarja támaszkodni egy elavult módszerre, amely a jövőben ismeretlen időpontban megszakad.
következő lépések
általában több fájlt szeretne kiírni párhuzamosan, de ritka esetekben, amikor egyetlen fájlt szeretne kiírni, a spark-daria writeSingleFile módszer segít.
próbálja meg a legjobb, hogy lezárja a komplex Hadoop fájlrendszer logika segítő módszerek, amelyek tesztelik elválasztva. A Hadoop fájlrendszer műveletek és a Spark kód kombinálása ugyanabban a módszerben túl összetetté teszi a kódot.