Pyspark.sql.DataFrameReader.csv()
Αυτή η μέθοδος χρησιμοποιείται για την ανάγνωση των δεδομένων από τα αρχεία CSV και την αποθήκευση τους στο PySpark DataFrame. Παίρνει τις επιλογές κατά την ανάγνωση του CSV στο DataFrame. Θα συζητήσουμε τις διάφορες επιλογές με παραδείγματα λεπτομερώς. Κατά τη διαβίβαση περισσότερων από ένα αρχείων CSV, είναι σημαντικό να μεταβιβάζονται τα ονόματα αρχείων με επέκταση σε μια λίστα που διαχωρίζεται από τον τελεστή κόμματος. Εάν διαβάζετε μόνο ένα αρχείο CSV, δεν χρειάζεται να δώσετε το όνομα αρχείου σε μια λίστα.
Σύνταξη:
Μεμονωμένο αρχείο - spark_app.read.csv('file.csv', επιλογές …)
Πολλαπλά αρχεία – spark_app.read.csv(['file1.csv','file2.csv',…],επιλογές…)
Μπορεί επίσης να είναι δυνατός ο διαχωρισμός των επιλογών και των ονομάτων αρχείων.
Εις μια σειρά – spark_app.read.options(options…).csv('file.csv')
Πολλαπλά αρχεία – spark_app.read.options(options…).csv(['file1.csv','file2.csv',…])
Εγκαταστήστε τη βιβλιοθήκη PySpark πριν εφαρμόσετε τα ακόλουθα παραδείγματα.
pip εγκατάσταση pyspark
Μετά την επιτυχή εγκατάσταση, μπορείτε να δείτε την έξοδο ως εξής:
Σενάριο 1: Ανάγνωση της κεφαλίδας αρχείου CSV
Ας δημιουργήσουμε ένα αρχείο CSV με το όνομα 'person_skill.csv' με 5 εγγραφές που εμφανίζονται παρακάτω και ας το φορτώσουμε στο PySpark DataFrame:
Η παράμετρος κεφαλίδας χρησιμοποιείται για τον καθορισμό των ονομάτων στηλών στο PySpark DataFrame. Παίρνει μια τιμή Boolean. Εάν είναι 'True', τα πραγματικά ονόματα στηλών που υπάρχουν στο αρχείο CSV καθορίζονται στο DataFrame, Διαφορετικά, καθορίζονται τα c0, c1, c2… και τα πραγματικά ονόματα στηλών θα είναι μια γραμμή. Είναι καλύτερο να ορίσετε την παράμετρο κεφαλίδας σε true.
Παράδειγμα 1: Κεφαλίδα = Σωστό
εισαγωγή pysparkαπό το pyspark.sql εισαγωγή SparkSession
linuxhint_spark_app = SparkSession.builder.appName( 'Συμβουλή Linux' .getOrCreate()
# Φορτώστε το csv με όνομα - person_skill.csv σε δεξιότητες με ετικέτες στηλών με κεφαλίδα
δεξιότητες = linuxhint_spark_app.read.csv( 'person_skill.csv' , επί κεφαλής = Αλήθεια)
# Εμφάνιση του DataFrame
skills.show()
Παραγωγή:
Εξήγηση:
Μπορούμε να δούμε ότι το PySpark DataFrame δημιουργείται από το αρχείο CSV με καθορισμένες στήλες και γραμμές.
Χρησιμοποιήστε την ακόλουθη εντολή για να ελέγξετε τις στήλες:
δεξιότητες.στήλες 
Παράδειγμα 2: Κεφαλίδα = Λάθος
εισαγωγή pysparkαπό το pyspark.sql εισαγωγή SparkSession
linuxhint_spark_app = SparkSession.builder.appName( 'Συμβουλή Linux' .getOrCreate()
# Φορτώστε το csv με όνομα - person_skill.csv σε δεξιότητες με ετικέτες στηλών χωρίς κεφαλίδα
δεξιότητες = linuxhint_spark_app.read.csv( 'person_skill.csv' , επί κεφαλής =Λάθος)
# Εμφάνιση του DataFrame
skills.show()
Παραγωγή:
Εξήγηση:
Μπορούμε να δούμε ότι το PySpark DataFrame δημιουργείται από το αρχείο CSV χωρίς υπάρχουσες στήλες.
Επίσης, οι υπάρχουσες στήλες αποθηκεύονται ως γραμμές στο PySpark DataFrame.
δεξιότητες.στήλες 
Χρησιμοποιώντας το Read.options.csv()
Τώρα, διαβάζουμε το αρχείο CSV χρησιμοποιώντας τη μέθοδο read.options.csv(). Εδώ, πρέπει να περάσουμε τις επιλογές όπως οριοθέτηση, κεφαλίδα κ.λπ. στις επιλογές ως ορίσματα και όνομα αρχείου στη csv(). Ας περάσουμε την παράμετρο κεφαλίδας ορίζοντας την σε 'True'.
Σενάριο 1:
εισαγωγή pysparkαπό το pyspark.sql εισαγωγή SparkSession
linuxhint_spark_app = SparkSession.builder.appName( 'Συμβουλή Linux' .getOrCreate()
# Χρήση read.options.csv()
δεξιότητες = linuxhint_spark_app.read. επιλογές ( επί κεφαλής =True).csv( 'person_skill.csv' )
# Εμφάνιση του DataFrame
skills.show()
Παραγωγή:
Σενάριο 2: Ανάγνωση του διαχωριστή αρχείων CSV
Η παράμετρος οριοθέτη παίρνει τον χαρακτήρα που χρησιμοποιείται για τον διαχωρισμό κάθε πεδίου. Παίρνει κόμμα (,) από προεπιλογή. Ας χρησιμοποιήσουμε το ίδιο αρχείο CSV που χρησιμοποιείται στο πρώτο σενάριο και ας περάσουμε το κόμμα (‘,’) ως οριοθέτη.
εισαγωγή pysparkαπό το pyspark.sql εισαγωγή SparkSession
linuxhint_spark_app = SparkSession.builder.appName( 'Συμβουλή Linux' .getOrCreate()
# Χρήση read.options.csv() με οριοθέτη μαζί με κεφαλίδα
δεξιότητες = linuxhint_spark_app.read. επιλογές ( επί κεφαλής =Αληθής,οριοθέτης= ',' ).csv( 'person_skill.csv' )
# Εμφάνιση του DataFrame
skills.show()
Παραγωγή:
Ανάγνωση πολλών αρχείων
Μέχρι τώρα, διαβάζαμε ένα μόνο αρχείο CSV. Ας δούμε πώς μπορείτε να διαβάσετε περισσότερα από ένα αρχεία CSV. Σε αυτό το σενάριο, οι σειρές σε πολλά αρχεία προστίθενται σε ένα μεμονωμένο PySpark DataFrame. Απλώς πρέπει να περάσουμε τα ονόματα αρχείων σε μια λίστα μέσα στη μέθοδο.
Παράδειγμα:
Ας έχουμε τα ακόλουθα αρχεία CSV με τα ονόματα 'person_skill.csv' και 'person_skill2.csv' με τα ακόλουθα δεδομένα:
Διαβάστε αυτά τα δύο αρχεία CSV και αποθηκεύστε τα σε ένα μόνο PySpark DataFrame.
εισαγωγή pysparkαπό το pyspark.sql εισαγωγή SparkSession
linuxhint_spark_app = SparkSession.builder.appName( 'Συμβουλή Linux' .getOrCreate()
# Φορτώστε 2 αρχεία csv με όνομα - person_skill.csv και person_skill2.csv σε δεξιότητες με ετικέτες στηλών με κεφαλίδα
skills = linuxhint_spark_app.read.csv([ 'person_skill.csv' , 'person_skill2.csv' ], σεπ= ',' , επί κεφαλής = Αλήθεια)
skills.show()
Παραγωγή:
Εξήγηση:
Το πρώτο CSV έχει 6 εγγραφές και το δεύτερο CSV 3 εγγραφές. Μπορούμε να δούμε ότι το πρώτο CSV φορτώνεται πρώτα στο DataFrame. Στη συνέχεια, φορτώνεται το δεύτερο CSV. Τέλος, το PySpark DataFrame έχει 9 εγγραφές.
συμπέρασμα
Η ανάγνωση του CSV στο PySpark DataFrame είναι αρκετά απλή με τη μέθοδο pyspark.sql.DataFrameReader.csv(). Μπορεί να είναι δυνατή η μεταβίβαση των παραμέτρων κεφαλίδας και οριοθέτησης σε αυτήν τη μέθοδο προκειμένου να καθοριστούν οι στήλες και η μορφή. Το PySpark υποστηρίζει επίσης την ανάγνωση πολλών αρχείων CSV ταυτόχρονα με τις παρεχόμενες μεθόδους μαζί με τις επιλογές τους. Σε αυτό το άρθρο, έχουμε δει τα παραδείγματα εξετάζοντας διαφορετικές επιλογές. Επίσης, έχουμε δει δύο τρόπους μετάδοσης των επιλογών στη μέθοδο.