Polber
commented
7 months ago I am unable to reproduce the error.
I used the following pipeline:
Schema SCHEMA = Schema.of(Schema.Field.of("event", Schema.FieldType.STRING));
Pipeline pipeline = Pipeline.create();
PCollection<Row> input = pipeline.apply(
Create.of(
Row.withSchema(SCHEMA).withFieldValue("event", "abc").build()
)
).setRowSchema(SCHEMA);
PCollection<Row> transformed = input.apply(
SqlTransform.query("select event as event_name, count(*) as c from PCOLLECTION group by event"));
transformed.apply(ParDo.of(new DoFn<Row, Object>() {
@ProcessElement
public void processElement(ProcessContext c) throws Exception {
System.out.println(c.element());
c.output(c.element());
}
})).setRowSchema(
Schema.of(
Schema.Field.of("event", Schema.FieldType.STRING),
Schema.Field.of("c", Schema.FieldType.INT64)
)
);
pipeline.run();Logs:
Mar 22, 2024 2:30:06 PM org.apache.beam.sdk.extensions.sql.impl.CalciteQueryPlanner convertToBeamRel
INFO: SQL:
SELECT `PCOLLECTION`.`event` AS `event_name`, COUNT(*) AS `c`
FROM `beam`.`PCOLLECTION` AS `PCOLLECTION`
GROUP BY `PCOLLECTION`.`event`
Mar 22, 2024 2:30:07 PM org.apache.beam.sdk.extensions.sql.impl.CalciteQueryPlanner convertToBeamRel
INFO: SQLPlan>
LogicalProject(event_name=[$0], c=[$1])
LogicalAggregate(group=[{0}], c=[COUNT()])
BeamIOSourceRel(table=[[beam, PCOLLECTION]])
Mar 22, 2024 2:30:07 PM org.apache.beam.sdk.extensions.sql.impl.CalciteQueryPlanner convertToBeamRel
INFO: BEAMPlan>
BeamCalcRel(expr#0..1=[{inputs}], proj#0..1=[{exprs}])
BeamAggregationRel(group=[{0}], c=[COUNT()])
BeamIOSourceRel(table=[[beam, PCOLLECTION]])
Mar 22, 2024 2:30:08 PM org.apache.beam.sdk.util.construction.Environments$JavaVersion forSpecification
WARNING: Unsupported Java version: 18, falling back to: 17Row:
Row:
event_name:abc
c:1If you give me more details on your pipeline, I can try to reproduce, but if you can model yours more like the one I posted above and see if that resolves the issue, that would be a good first step.
As far as...
Event_name column is correct, but why C0? and not c?
This is to avoid name collision. In your query, you select a column c and also create a column c by aggregating count(*), so this is working as expected.
brachipa
kennknowles
liferoad
What happened?
We have many
SqlTransformin our pipeline, running SQL query and sink output into external system (Kinesis), we found that in some cases our SQL results aren't matching the SQL we provided. for example the bellow simple SQL:returns output with fields "event" and "c" ignoring the alias "event_name"
It happens without any reason, and by applying some simple change (e.g removing one column from group by ), the query stopped outputting data into the "event_name" only to "event" column and it caused us to huge data loss, since all our metrics are based on event_name, and this one become always null.
We need to understand what causes aliases to be ignored, since we are using this pattern in most queries and in different fields. all are in production critical system.
logs for the above query:
But the row we get: row.toString():
The pipeline steps look like:
Adding more fields to the query can sometimes help, but this is not always the case, and we turned into situation that we can't edit our queries since we don't know what the impact will be.
Query that works:
Logs:
Row looks like:
Event_name column is correct, but why C0? and not c?
Also, it is not always number of group by fields, if I send the exact same query, only with real fields (not a,b,c) it is again stop sending event_name only event.
Can you explain this strange bug?
Issue Priority
Priority: 1 (data loss / total loss of function)
Issue Components