Как считать датасет в Spark?

Все, мы импортировали все, что нужно, создали сессию, нашли датасет, готовы работать!

Самые популярные способы:
⭐ csv

df = (
  spark.read
  .option('header', True)
  .option('delimiter', ';')
  .option('inferSchema', False)
  .csv(path)
)

val df = spark.read
  .option("header", "true")
  .option("delimiter", ";")
  .option("inferSchema", "false")
  .csv(path)

На месте path может быть 'file.csv', несколько файлов 'file1.csv,file2.csv' или вся папка целиком 'myfiles/csv_data'.

Из опций основные:

🐱 header - заголовок
🐱 delimiter - разделитель полей
🐱 inferSchema - автоматический определитель типов полей

На больших данных использовать inferSchema опасно!

Почему? Потому что спарк пойдет сканировать каждую строчку, чтобы определить корректный тип. Допустим, есть колонка age, заполненная числами типа int, но в сааамом конце у нас все поехало и попало число типа double. В конечном итоге столбец будет тоже иметь тип double, но для этого мы просканировали все на всякий случай. И так с каждым полем. Ну такое😕

Поэтому надо самим задать схему и указать ее при чтении файла:

table_schema = StructType([
  StructField('name', StringType(), True),  # True - допустимы ли null
  StructField('age', IntegerType(), True),
  ...
])

df = (
  spark.read
  .option(...)
  .schema(table_schema)
  .csv(path)
)

StructType - класс для структуры датафрейма. StructField - для поля. Лежат в spark.sql.types.

⭐json, parquet, hive-таблица

df = spark.read.json(path)
df = spark.read.parquet(path)
df = spark.table('db_schema.table_name')

У нас есть HDFS (Hadoop Distributed File System) - это распределенная файловая система, в которой обычно хранятся большие данные. Там они лежат в виде файликов внутри папочек. А Hive - это обертка, чтобы мы могли читать данные из файликов через sql-запросы🐝

#spark