Уже раза три тут это обсуждали

customSchema в п-мщ

И createTableOptions

Это ток на чтение

Сюда как раз engine прописывается

Чёт не получается😢

def writeToCh4(data: DataFrame, jdbcUrl: String, password: String, username: String, database: String, tableName: String): Unit = {
  val props = new java.util.Properties()

  val createTableColumnTypes = "id VARCHAR(1024), stdVersion INTEGER"

  val createTableOptions = s"CREATE TABLE $database.$tableName (id UUID, param2 Int32) ENGINE = MergeTree() ORDER BY (id)"
  
  data.write
    .mode(SaveMode.Append)
    .option("createTableColumnTypes", createTableColumnTypes)
    .option("createTableOptions", createTableOptions)
    .option("driver", "ru.yandex.clickhouse.ClickHouseDriver")
    .option("password", password)
    .option("connection", jdbcUrl)
    .option("user", username)
    .jdbc(jdbcUrl, tableName, props)

}

val arr = Array(
    ("asfaf1", 1),
    ("bbbb2", 2)
)
val df1 = sc.parallelize(arr).toDF("id", "param2")

writeToCh4(df1, jdbcUrl, password, username, "<DATABASE>", "<TABLENAME>")

ru.yandex.clickhouse.except.ClickHouseException: ClickHouse exception, code: 62, host: <HOST>, port: <PORT>; Code: 62, e.displayText() = DB::Exception: Syntax error: failed at position 65: NOT NULL) CREATE TABLE <DATABASE>.<TABLENAME> (id UUID, param2 Int32) ENGINE = MergeTree() ORDER BY (id). Expected one of: CODEC, ALIAS, TTL, ClosingRoundBracket, Comma, DEFAULT, MATERIALIZED, COMMENT, token (version 20.3.8.53 (official build))

def writeToCh4(data: DataFrame, jdbcUrl: String, password: String, username: String, database: String, tableName: String): Unit = {
  val props = new java.util.Properties()

  val createTableColumnTypes = "id VARCHAR(1024), stdVersion INTEGER"

  val createTableOptions = s"CREATE TABLE $database.$tableName (id UUID, param2 Int32) ENGINE = MergeTree() ORDER BY (id)"
  
  data.write
    .mode(SaveMode.Append)
    .option("createTableColumnTypes", createTableColumnTypes)
    .option("createTableOptions", createTableOptions)
    .option("driver", "ru.yandex.clickhouse.ClickHouseDriver")
    .option("password", password)
    .option("connection", jdbcUrl)
    .option("user", username)
    .jdbc(jdbcUrl, tableName, props)

}

val arr = Array(
    ("asfaf1", 1),
    ("bbbb2", 2)
)
val df1 = sc.parallelize(arr).toDF("id", "param2")

writeToCh4(df1, jdbcUrl, password, username, "<DATABASE>", "<TABLENAME>")

ru.yandex.clickhouse.except.ClickHouseException: ClickHouse exception, code: 62, host: <HOST>, port: <PORT>; Code: 62, e.displayText() = DB::Exception: Syntax error: failed at position 65: NOT NULL) CREATE TABLE <DATABASE>.<TABLENAME> (id UUID, param2 Int32) ENGINE = MergeTree() ORDER BY (id). Expected one of: CODEC, ALIAS, TTL, ClosingRoundBracket, Comma, DEFAULT, MATERIALIZED, COMMENT, token (version 20.3.8.53 (official build))

Доку читать нужно внимательно. Или на худой конец в исходники посмотреть

Гайз, есть сервис из Kafka - источник данных, Spark Streaming - обработка, MongoDB - хранилище. Я правильно понимаю, что минимальный состав по хостам - 3 штука, на каждом из которых стоит Kafka, Spark (1 из них master и worker одновременно) и MongoDB. Интересует именно минимальное кол-во хостов и то, что каждый хост тащит каждый из 3 компонентов, логично ли это\

Ну можно и на одном все поднять:)

Этот этап пройден :) Дальше нужно двигаться для обеспечения устойчивости, и хотелось бы понять, как правильно. Может кто ткнет ссылкой на тему

А у вас прод на железе и виртуалках? Если виртуалки то лучше компоненты не смешивать. Просто то, что вы описываете очень индивидуальный процесс. И зависит от того какие именно компоненты должны быть более устойчивы

У разных заказчиков по разному. Пока на вирталках, они справляются. Но с ростом кол-ва входящих будет переход на реальное железо. Тут важный момент, что это сервис мониторинга (не кртичный) и сейчас работает без Кафки совсем. Т.е. кратковременной упавший сервак упавшим не считается 😁 и это норм. Сейчас нужно исследовать уже работу с Кафкой, проверить производительность и наконец-то устойчивость. И не очень понятно, с какой минимум конфигурации начать

> kafka
> проверить производительность
> проверить устойчивость
> 2020

Делать кафку ха и монгу разные вещи.
Но при том и том я бы сначала использовал здравый смысл. Потом entry level гайды

к тому же кафка это не только сама кафка но и зукипер, у которого своя история

Это смотря какие требования у вас к надёжности.  Например , если у вас 2 дата центра и вам нужна система , что переживает условное падение одного дата цента , то вам нужно 4 брокера для Кафки и зк

ну и еще настроить репликацию меж дц

Zk нечётное количество должно быть, для консенсуса

Просто тема достаточно глубокая, кроме кафки нужно ещё ha для ваших сервисов, а это либо дублирование с внешним балансировщиком, либо какой-то оркестратор, но для него тоже ha должно быть

вообще то может быть и четное. Кворум в ЗК просто через большинство. см https://github.com/apache/zookeeper/blob/efe83baeaa6715c32383a383f33f70522c1b7614/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/flexible/QuorumMaj.java#L136 . Но в слуачае 2 ДС соглашусь.