Но хорошо что разобрались

Интересно, как это работает )

Блин. Теперь dataset не джойниться
<console>:51: error: not enough arguments for method apply

Народ а вот такой вопрос.
У меня после join далее формируется запрос на сохранение (к примеру) или дальнейшие преобразования. И вот что я вижу: количество тасков после join из источника данных для всех остальных stage одинаковое и равно 200. Я так понимаю, это задано дефолтным значение для reduce а как то управлять этим можно динамически?

Увы. Задано изначально spark.sql.shuffle.partitions, и либо его править, либо repartition/coalesce.

Репартишн вызовет оверхед для шаффл. Не очень хочется

Понятно. Я думал этим можно как то управлять в динамике.

А кстати есть какие бест практис по определению
spark.sql.shuffle.partitions

Всем привет
Я никак не могу понять как отфильтровать Dataset с case классами с Options

Т.е.  допустим есть

case class Person(name: String, age: Option[Int])
case class User(id: Int, person: Option[Person])

val users = Dataset[User]()

И мне надо отфильтровать SeDatasetq[Cl1] где age > 20
Ну а если person None или age None, то соответственно отбрасывать, т.е. функция в filter должна возвращать false

Как это сделать без

users.filter{ item =>
  item.person match {
    case Some(person) => person.age match {
      case Some(age) => age > 20
      case None => false
    }
    case None => false
  }
}

Как наисать это без match case? Это возможно написать с помощью flatMap ?

case class Person(name: String, age: Option[Int])
case class User(id: Int, person: Option[Person])

val users = Dataset[User]()

users.filter{ item =>
  item.person match {
    case Some(person) => person.age match {
      case Some(age) => age > 20
      case None => false
    }
    case None => false
  }
}

_.person.flatMap(_.age).fold(false){age => if (age > 20) true else false}

👍
Спс

Даже можно _ > 20 в конце

Да да

Всем добрый день. Хочу услышать ваше мнение и жду ссылки на статьи😁. В данный момент используем cassandra для хранения данных (для построения отчетности я выгружают необходимые таблицы за весь временной промежуток на комп и произвожу манипуляции в питоне, минус в том, что эти таблицы по нескольку часов выгружаются). Так же используем связку Elasticksearc+Kibana для оперативной отчетности, но данные не за весь период хранения. Хотим перенести нашу БД (некоторые таблицы содержат по 10М записей) на Greenplum или Clickhouse. Вопросы: 1. куда лучше перенести? 2. нужно ли изменять структуру БД (добавлять какие-нибудь данные, менять отношения - вместо 1 к 1 менять 1 ко многим)? 3. исходя из чего выбрать промежуток для записи данных? 4. как перелить данные(spark или еще что то использовать, может самим написать)? Буду рад любым идеям и замечаниям😁

А почем инкремент не выделяете?
Создайте рядом таблицу индекс выгруженных id и перед стартом формируйте список id на выгрузку.

Кассандру держат для поддержки штанов(большого кол-ва мелких инсертов), применять под эту задачу MPP такое себе решение

мне посоветовали использовать greenplum (т.к. не миллиарды записей) + cdc для проверки когда нужно считывать данные из с* и писать в greenplum. структуру пока оставить ту же. копирование данных из c*  в greenplum производить своим скриптом. что думаете?

greenplum вам зачем, простым postgresql не обойтись?

Вышел Spark 3.1.1. Для меня основные фишки этого релиза - это улучшения в рамках Project Zen (Make PySpark great again и все такое).
https://spark.apache.org/releases/spark-release-3-1-1.html

Две жутко бесячих особенности PySpark (нечитаемые трейсбеки и нечитаемая документация) пофиксили. Документация реально выглядит лучше и уже не грузится одним огромным HTML-ем с кучей якорей.
https://issues.apache.org/jira/browse/SPARK-33407
https://issues.apache.org/jira/browse/SPARK-31851

Завезли type hints. То, что долго и относительно успешно делал проект pyspark-stubs. https://issues.apache.org/jira/browse/SPARK-32681

В Spark SQL много улучшений в shuffled hash join https://issues.apache.org/jira/browse/SPARK-32461