Я пытаюсь создать потоковый конвейер с использованием apache-beam, который читает предложения из google pub / sub и записывает слова в таблицу Bigquery.
Я использую 0.6.0
версию apache-beam.
Следуя примерам, я сделал это:
public class StreamingWordExtract {
/**
* A DoFn that tokenizes lines of text into individual words.
*/
static class ExtractWords extends DoFn<String, String> {
@ProcessElement
public void processElement(ProcessContext c) {
String[] words = ((String) c.element()).split("[^a-zA-Z']+");
for (String word : words) {
if (!word.isEmpty()) {
c.output(word);
}
}
}
}
/**
* A DoFn that uppercases a word.
*/
static class Uppercase extends DoFn<String, String> {
@ProcessElement
public void processElement(ProcessContext c) {
c.output(c.element().toUpperCase());
}
}
/**
* A DoFn that uppercases a word.
*/
static class StringToRowConverter extends DoFn<String, TableRow> {
@ProcessElement
public void processElement(ProcessContext c) {
c.output(new TableRow().set("string_field", c.element()));
}
static TableSchema getSchema() {
return new TableSchema().setFields(new ArrayList<TableFieldSchema>() {
// Compose the list of TableFieldSchema from tableSchema.
{
add(new TableFieldSchema().setName("string_field").setType("STRING"));
}
});
}
}
private interface StreamingWordExtractOptions extends ExampleBigQueryTableOptions, ExamplePubsubTopicOptions {
@Description("Input file to inject to Pub/Sub topic")
@Default.String("gs://dataflow-samples/shakespeare/kinglear.txt")
String getInputFile();
void setInputFile(String value);
}
public static void main(String[] args) {
StreamingWordExtractOptions options = PipelineOptionsFactory.fromArgs(args)
.withValidation()
.as(StreamingWordExtractOptions.class);
options.setBigQuerySchema(StringToRowConverter.getSchema());
Pipeline p = Pipeline.create(options);
String tableSpec = new StringBuilder()
.append(options.getProject()).append(":")
.append(options.getBigQueryDataset()).append(".")
.append(options.getBigQueryTable())
.toString();
p.apply(PubsubIO.read().topic(options.getPubsubTopic()))
.apply(ParDo.of(new ExtractWords()))
.apply(ParDo.of(new StringToRowConverter()))
.apply(BigQueryIO.Write.to(tableSpec)
.withSchema(StringToRowConverter.getSchema())
.withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
.withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND));
PipelineResult result = p.run();
}
У меня ошибка рядом:
apply(ParDo.of(new ExtractWords()))
потому что предыдущий apply
не возвращает String
, а Object
Я полагаю, что проблема в типе, возвращенном PubsubIO.read().topic(options.getPubsubTopic())
. Тип - PTransform<PBegin, PCollection<T>>
вместо PTransform<PBegin, PCollection<String>>
.
Как правильно читать из Google Pub / Sub с помощью apache-beam?