Описание тега structured-streaming

У меня есть таблица exchangeRates, которая обновляется в пакетном режиме один раз в неделю. Это должно использоваться другими пакетными и потоковыми заданиями в разных кластерах - поэтому я хочу сохранить это как постоянную общую таблицу для общего …

Я использую структурированный поток с Spark 2.2. Мы используем Kafka в качестве нашего источника и используем контрольные точки для восстановления после сбоев, и e2e гарантирует только один раз. Я хотел бы получить дополнительную информацию о том, к…

У меня есть потоковый запрос (Spark Structured Streaming), который получает данные из темы Kafka (два раздела), например: val df = spark .readStream .format("kafka") .option("kafka.bootstrap.servers", "172.29.57.25:9092,172.29.57.30:9092") .option("…

Я выполняю запрос на SparkSession который бросает Object is not an instance of declaring classниже приведен код, после которого Dataset<Row> results = spark.sql("SELECT t1.someCol FROM table1 t1 join table2 t2 on t1.someCol=t2.someCol"); resul…

Spark 2.2 представил структурированный потоковый источник Kafka. Как я понимаю, он использует каталог контрольных точек HDFS для хранения смещений и гарантии доставки сообщений "точно один раз". Но старые доки (такие как https://blog.cloudera.com/bl…

Поэтому я пытаюсь настроить поток из Kafka, а затем добавить столбцы из таблицы в этот фрейм данных на основе соответствующего идентификатора в сообщении и таблице Kafka. У меня есть два отдельных файла, первый вызывает метод в streamKafka_ASH для н…

Код сервера: import java.net.ServerSocket; import java.net.Socket; import java.nio.charset.Charset; import java.util.Random; public class Socket_server { public static void main(String[] args) throws Exception { ServerSocket sc = new ServerSocket(99…

Я пытаюсь использовать структурированную потоковую передачу в Spark 2.1.1 для чтения из Kafka и декодирования закодированных сообщений Avro. У меня есть UDF, определенный в соответствии с этим вопросом. val sr = new CachedSchemaRegistryClient(conf.k…

В настоящее время у меня есть следующий df +-------+--------------------+-----+ | key| created_at|count| +-------+--------------------+-----+ |Bullish|[2017-08-06 08:00...| 12| |Bearish|[2017-08-06 08:00...| 1| +-------+--------------------+-----+ Я…

Насколько точны оценки водяных знаков при обработке потока в пучке Apache или в потоке искры. Мой источник данных - файлы из gcs/s3, но я использую время события, связанное с каждым событием, как метку времени для функции управления окнами. Любые ид…

В моем случае у меня есть файлы CSV, постоянно загружаемые в HDFS. Как только новый файл будет загружен, я хотел бы обработать новый файл с помощью Spark SQL (например, вычислить максимум поля в файле, преобразовать файл в parquet). т.е. у меня есть…

Моя команда сейчас вступает в сферу структурированного потокового вещания. Я относительно новичок в структурированном потоке. У меня есть требование с Источник - CSVРаковина - JSON Подробности Env: Кластер: Искра 2.2.1Язык программирования: ScalaИнс…

Я следовал этим шагам, чтобы настроить Prometheus, Graphite Exporter и Grafana для построения метрик для Spark 2.2.1, в которой работает структурированная потоковая передача. Метрики коллекции на этом посте довольно устарели; и не включает в себя ка…

Мы можем видеть StateStoreRestoreExec следующим образом. case class StateStoreRestoreExec( keyExpressions: Seq[Attribute], stateId: Option[OperatorStateId], child: SparkPlan) extends UnaryExecNode with StateStoreReader { override protected def doExe…

У меня есть простая работа с триггером =15 секунд, Source=Kafka и Sink=S3. Можно ли узнать, сколько времени заняло скачивание сообщений с Кафки? Или, скажем, если у меня была Sink=Console, она возвращает данные о драйвере, можно ли найти сколько вре…

вопрос Я использую Spark, чтобы присоединиться к содержимому файлов CSV. После первого соединения с потоковым CSV-файлом я могу записать поток без ошибок и увидеть, что было выведено много строк, но как только я пытаюсь записать свой второй поток, я…

Я использую Spark 2.1.1. У меня есть следующее DataSet<Row> DS1; name | ratio | count // column names "hello" | 1.56 | 34 (ds1.isStreaming дает true) и я пытаюсь создать DataSet<String> ds2. другими словами, когда я пишу в кафку, я хочу …

Я пытаюсь запустить простой пример структурированной потоковой передачи с веб-сайта Apache Spark. Я подаю свое заявление в Spark следующим образом: spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.2.0 spark_structured_streaming.p…

Я использую Структурированную Потоковую передачу Spark 2.1 (которая внутренне использует банки Jive 1.2.1). Я пытаюсь разработать ForEachWriter для Hive для записи потоковых данных в Hive через JDBC-коннектор. На моем кластере у меня Hive 1.1.0. Есл…

Я пытаюсь получить ряд прочитанных и записанных записей в структурированном потоке. я использую SparkListener.onTaskEnd() для захвата входных и выходных метрик, но кажется, что количество записанных записей (taskEnd.taskMetrics().outputMetrics().rec…