Описание тега apache-beam-internals

Вопросы с тегом
2 ответа

TextIO.Read().From() против TextIO.ReadFiles() поверх withHintMatchesManyFiles()

В моем случае получения набора подходящих файловых шаблонов от Kafka, PCollection<String> filepatterns = p.apply(KafkaIO.read()...); Здесь каждый шаблон может соответствовать до 300+ файлов. Q1. Как я могу использоватьTextIO.Read() чтобы сопос…
01 июн '20 в 08:09
1 ответ

Получение так много предупреждений при использовании List с настраиваемым Java-классом POJO в apache beam java

Я новичок в Apache beam, я использую луч Apache и в качестве бегуна использую поток данных в GCP. Я получаю следующую ошибку при выполнении конвейера. coder of type class org.apache.beam.sdk.coders.ListCoder has a #structuralValue method which does …
12 июн '20 в 10:58
2 ответа

Ошибка при запуске Apache Beam Python SplittableDoFn

Ошибка при попытке pubsub io> splittable dofn RuntimeError: узел преобразования AppliedPTransform(ParDo(TestDoFn)/ProcessKeyedElements/GroupByKey/GroupByKey, _GroupByKeyOnly) не был заменен должным образом. Может ли кто-нибудь помочь мне с проверкой…
04 авг '20 в 20:40
0 ответов

Apache Beam Python SDK устанавливает водяной знак вручную с помощью тела события

Пример использования: создание неограниченной коллекции Pcollection из темы pubsub Водяной знак будет иметь вид "Идеальный водяной знак", доступный в теле сообщения. Источник очереди pubsub будет определять водяной знак Message { event_time timestam…
03 авг '20 в 18:53
2 ответа

Знать количество потоков, запущенных в прямом бегуне Apache Beam

У меня есть программа Apache Beam в Java, работающая с прямым бегуном. Луч Apache использует потоки для достижения распределенной обработки. Во время выполнения, как я могу узнать количество потоков, порожденных лучом apache? Как я могу установить м…
10 июл '20 в 09:16
1 ответ

Компромисс между скоростью и памятью при разделении Apache Beam PCollection на две части

У меня есть PCollection, где каждый элемент является ключом, кортеж значений выглядит следующим образом: (key, (value1,..,value_n) ) Мне нужно разделить эту коллекцию PCollection на две ветви обработки. Как всегда, мне нужно, чтобы весь конвейер раб…
04 дек '20 в 15:10
1 ответ

Загрузка файла в БД с помощью Apache Beam

I have scenario as below: I need to load file into Database. But before that I have to verify data is present into Database based on file data. Now suppose I have 5 records in a file then I have to check 5 times into database for separate records. …
12 мар '21 в 14:47
1 ответ

beam.Create() со списком dicts работает очень медленно по сравнению со списком строк

Я использую Dataflow для обработки шейп-файла с примерно 4 миллионами функций (всего около 2 ГБ) и загрузки геометрии в BigQuery, поэтому перед запуском конвейера я извлекаю функции шейп-файла в список и инициализирую конвейер, используя. Есть два с…
30 дек '20 в 09:35
1 ответ

Я вижу масштабирование луча apache с # из файлов CSV легко, но как насчет # строк в одном CSV?

В настоящее время я читаю эту статью и документы apache beam https://medium.com/@mohamed.t.esmat/apache-beam-bites-10b8ded90d4c Все, что я читал, касается N файлов. В нашем случае мы каждый раз получаем событие pubsub для ОДНОГО нового файла, чтобы …
18 апр '21 в 07:28
1 ответ

Разница между оконной и оконной балкой apache

В чем разница между панелью и окном? Поступающие элементы группируются в окна. Тогда что содержит панель? Я взял следующий код из документации по лучу .of(new DoFn<String, String>() { public void processElement(@Element String word, PaneInfo p…
02 май '21 в 10:47
1 ответ

Эквивалент передела в луче apache

В Spark, если нам нужно перетасовать данные, мы можем использовать повторное разбиение фрейма данных. Как можно сделать то же самое в Apache Beam для коллекции pcollection? В pyspark, new_df = df.repartition(4)
02 май '21 в 14:10
2 ответа

Я получаю эту ошибку. Получение SEVERE Channel ManagedChannelImpl{logId=1, target=bigquerystorage.googleapis.com:443} не было завершено должным образом

Я создал сценарий луча для получения данных из kafka и отправки их в BigQuery с помощью Apache Beam. На данный момент я использую java-direct-runner, и мне просто нужно отправить данные в свой большой запрос. Это мой код: - package com.knoldus.secti…
19 май '21 в 19:09
0 ответов

Spark Runner Apache Beam на AWS lambda

Я использую SparkRunner Apache Beam для преобразования некоторых данных (SparkRunner позволяет Apache Beam запускать Apache Spark Engine поверх своего конвейера). Приложение отлично работает на моем локальном компьютере. Я создаю банку и использую -…
18 июн '21 в 11:00
0 ответов

Луч Apache Join.leftOuterJoin, как передать пустой TableRow?

Я хочу выполнить leftOuterJoin на 2 таблицах BigQuery в Apache Beam (JAVA Sdk). Читаю таблицы ( leftTableCollection & rightTableCollection) и имеют тип PCollection<TableRow>. Преобразовал их в форму PCollection<KV<String, TableRow&gt…
31 июл '21 в 15:15
1 ответ

Beam DirectRunner Calcite не может указать имя

Я запускаю упрощенную версию этого руководства по лучу , но использую DirectRunner на моем локальном компьютере. import apache_beam as beam from apache_beam.transforms.sql import SqlTransform import os with beam.Pipeline() as p: rows = (p | beam.Cre…
01 сен '21 в 21:35
1 ответ

Apache Beam обновляет значения текущей строки на основе значений из предыдущей строки

Значения обновления Apache Beam основаны на значениях из предыдущей строки Я сгруппировал значения из файла CSV. Здесь, в сгруппированных строках, мы находим несколько пропущенных значений, которые необходимо обновить на основе значений из предыдуще…
02 ноя '21 в 00:46
1 ответ

Java Apache Beam: как добавить новые строки в коллекцию PCollection

В следующем CSV мне нужно добавить для него новые значения строк. Код Java: public static void main(String[] args) throws IOException { final File schemaFile = new File("src/main/resources/addRow/schema_transform.avsc"); File csvFile = new File("src…
11 ноя '21 в 03:41
1 ответ

Извлечение записей об ошибках при вставке в таблицу db с использованием JDBCIO apache beam в java

Я создаю в памяти коллекцию PCollection и записываю ее в postgres sql. теперь, когда я вставляю данные в таблицу, некоторые записи могут вызывать исключение и не будут вставлены. как извлечь такие неудачные записи вставки, когда я запускаю конвейер?…
22 ноя '21 в 16:43
0 ответов

Apache Beam — несколько коллекций P — проблема с преобразованием фрейма данных

Я запускаю приведенный ниже образец в луче apache. def transformdf(a, b): a["addr"] = "addr-common" return a p = beam.Pipeline(options=pipeline_options) data1 = [Row(id=1, name="abc"), Row(id=2, name="def"), Row(id=3, name="ghi")] pcol1 = (p | "Crea…
01 фев '22 в 11:13
0 ответов

Apache Beam Python — преобразование SQL с именованной проблемой PCollection

Я пытаюсь выполнить приведенный ниже код, в котором я использую Named Tuple для PCollection и преобразование SQL для выполнения простого выбора. По ссылке на видео (4:06): https://www.youtube.com/watch?v=zx4p-UNSmrA. Вместо использования PCOLLECTION…
01 фев '22 в 15:16