SQL for Avro GenericRecords on Parquet
I have tried to persist Generic Avro records in a parquet file and then read it via ParquetTablesource – using SQL. Seems that the SQL I not executed properly ! The persisted records are : Id , type 333,Type1 22,Type2 333,Type1 22,Type2 333,Type1 22,Type2 333,Type1 22,Type2 333,Type1 22,Type2 333,Type1 22,Type2 While SQL of SELECT id ,recordType_ FROM ParquetTable - return the above ( which is correct) Running : "SELECT id ,recordType_ FROM ParquetTable where recordType_='Type1' " Will result in : 333,Type1 22,Type1 333,Type1 22,Type1 333,Type1 22,Type1 333,Type1 22,Type1 333,Type1 22,Type1 333,Type1 22,Type1 As if the equal sign is assignment and not equal … am I doing something wrong ? is it an issue of Generic record vs SpecificRecords ?
Re: SQL for Avro GenericRecords on Parquet
Hi Hanan, Thanks for reporting the issue. Would you please attach your test code here? I may help to investigate. Best Regards Peter Huang On Mon, Nov 18, 2019 at 2:51 AM Hanan Yehudai wrote: > I have tried to persist Generic Avro records in a parquet file and then > read it via ParquetTablesource – using SQL. > Seems that the SQL I not executed properly ! > > The persisted records are : > Id , type > 333,Type1 > 22,Type2 > 333,Type1 > 22,Type2 > 333,Type1 > 22,Type2 > 333,Type1 > 22,Type2 > 333,Type1 > 22,Type2 > 333,Type1 > 22,Type2 > > While SQL of SELECT id ,recordType_ FROM ParquetTable - return the > above ( which is correct) > Running : "SELECT id ,recordType_ FROM ParquetTable where > recordType_='Type1' " > Will result in : > 333,Type1 > 22,Type1 > 333,Type1 > 22,Type1 > 333,Type1 > 22,Type1 > 333,Type1 > 22,Type1 > 333,Type1 > 22,Type1 > 333,Type1 > 22,Type1 > > As if the equal sign is assignment and not equal … > > am I doing something wrong ? is it an issue of Generic record vs > SpecificRecords ? > > >
RE: SQL for Avro GenericRecords on Parquet
chm); ParquetTableSource parquetTableSource = ParquetTableSource.builder() .path(path.getPath()) .forParquetSchema(nestedSchema) .build(); return parquetTableSource; } @Test public void testScan() throws Exception { ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); BatchTableEnvironment batchTableEnvironment = BatchTableEnvironment.create(env); ParquetTableSource tableSource = createParquetTableSource(testPath); batchTableEnvironment.registerTableSource("ParquetTable", tableSource); Table tab = batchTableEnvironment.sqlQuery("select id,recordType_ from ParquetTable where id > 22 "); DataSet result = batchTableEnvironment.toDataSet(tab, Row.class); result.print(); } } From: Peter Huang Sent: Monday, November 18, 2019 7:22 PM To: dev Cc: user@flink.apache.org Subject: Re: SQL for Avro GenericRecords on Parquet Hi Hanan, Thanks for reporting the issue. Would you please attach your test code here? I may help to investigate. Best Regards Peter Huang On Mon, Nov 18, 2019 at 2:51 AM Hanan Yehudai mailto:hanan.yehu...@radcom.com>> wrote: I have tried to persist Generic Avro records in a parquet file and then read it via ParquetTablesource – using SQL. Seems that the SQL I not executed properly ! The persisted records are : Id , type 333,Type1 22,Type2 333,Type1 22,Type2 333,Type1 22,Type2 333,Type1 22,Type2 333,Type1 22,Type2 333,Type1 22,Type2 While SQL of SELECT id ,recordType_ FROM ParquetTable - return the above ( which is correct) Running : "SELECT id ,recordType_ FROM ParquetTable where recordType_='Type1' " Will result in : 333,Type1 22,Type1 333,Type1 22,Type1 333,Type1 22,Type1 333,Type1 22,Type1 333,Type1 22,Type1 333,Type1 22,Type1 As if the equal sign is assignment and not equal … am I doing something wrong ? is it an issue of Generic record vs SpecificRecords ?
Re: SQL for Avro GenericRecords on Parquet
t;}"; > > > > private static final AvroSchemaConverter SCHEMA_CONVERTER = new > AvroSchemaConverter(); > > private static Schema schm = new Schema.Parser().parse(avroSchema); > > private static Path testPath; > > > > > > public ParquetTestCase() { > > super(TestExecutionMode.COLLECTION); > > } > > > > > > @BeforeClass > > public static void setup() throws Exception { > > > > GenericRecordBuilder genericRecordBuilder = new > GenericRecordBuilder(schm); > > > > > > List recs = new ArrayList<>(); > > for (int i = 0; i < 6; i++) { > > GenericRecord gr = genericRecordBuilder.set("timestamp_edr", > System.currentTimeMillis() / 1000).set("id", 333L).set("recordType_", > "Type1").build(); > > recs.add(gr); > > GenericRecord gr2 = genericRecordBuilder.set("timestamp_edr", > System.currentTimeMillis() / 1000).set("id", 22L).set("recordType_", > "Type2").build(); > > recs.add(gr2); > > } > > > > testPath = new Path("/tmp", UUID.randomUUID().toString()); > > > > > > ParquetWriter writer = > AvroParquetWriter.builder( > > new > org.apache.hadoop.fs.Path(testPath.toUri())).withSchema(schm).build(); > > > > for (IndexedRecord record : recs) { > > writer.write(record); > > } > > writer.close(); > > } > > > > > > private ParquetTableSource createParquetTableSource(Path path) throws > IOException { > > MessageType nestedSchema = SCHEMA_CONVERTER.convert(schm); > > ParquetTableSource parquetTableSource = > ParquetTableSource.builder() > > .path(path.getPath()) > > .forParquetSchema(nestedSchema) > > .build(); > > return parquetTableSource; > > } > > > > @Test > > public void testScan() throws Exception { > > ExecutionEnvironment env = > ExecutionEnvironment.getExecutionEnvironment(); > > > > BatchTableEnvironment batchTableEnvironment = > BatchTableEnvironment.create(env); > > ParquetTableSource tableSource = > createParquetTableSource(testPath); > > batchTableEnvironment.registerTableSource("ParquetTable", > tableSource); > > > > Table tab = batchTableEnvironment.sqlQuery("select > id,recordType_ from ParquetTable where id > 22 "); > > > > DataSet result = batchTableEnvironment.toDataSet(tab, > Row.class); > > > > result.print(); > > > > } > > > > > > } > > > > > > *From:* Peter Huang > *Sent:* Monday, November 18, 2019 7:22 PM > *To:* dev > *Cc:* user@flink.apache.org > *Subject:* Re: SQL for Avro GenericRecords on Parquet > > > > Hi Hanan, > > > > Thanks for reporting the issue. Would you please attach your test code > here? I may help to investigate. > > > > > > > > Best Regards > > Peter Huang > > > > On Mon, Nov 18, 2019 at 2:51 AM Hanan Yehudai > wrote: > > I have tried to persist Generic Avro records in a parquet file and then > read it via ParquetTablesource – using SQL. > Seems that the SQL I not executed properly ! > > The persisted records are : > Id , type > 333,Type1 > 22,Type2 > 333,Type1 > 22,Type2 > 333,Type1 > 22,Type2 > 333,Type1 > 22,Type2 > 333,Type1 > 22,Type2 > 333,Type1 > 22,Type2 > > While SQL of SELECT id ,recordType_ FROM ParquetTable - return the > above ( which is correct) > Running : "SELECT id ,recordType_ FROM ParquetTable where > recordType_='Type1' " > Will result in : > 333,Type1 > 22,Type1 > 333,Type1 > 22,Type1 > 333,Type1 > 22,Type1 > 333,Type1 > 22,Type1 > 333,Type1 > 22,Type1 > 333,Type1 > 22,Type1 > > As if the equal sign is assignment and not equal … > > am I doing something wrong ? is it an issue of Generic record vs > SpecificRecords ? > >
Re: SQL for Avro GenericRecords on Parquet
t; >> " \"fields\": [\n" + >> >> "{ \"default\": null, \"name\": \"timestamp_edr\", >> \"type\": [ \"null\", \"long\" ]},\n" + >> >> "{ \"default\": null, \"name\": \"id\", \"type\": [ >> \"null\", \"long\" ]},\n" + >> >> "{ \"default\": null, \"name\": \"recordType_\", >> \"type\": [ \"null\", \"string\"]}\n" + >> >> " ],\n" + >> >> " \"schema_id\": 1,\n" + >> >> " \"type\": \"record\"\n" + >> >> "}"; >> >> >> >> private static final AvroSchemaConverter SCHEMA_CONVERTER = new >> AvroSchemaConverter(); >> >> private static Schema schm = new Schema.Parser().parse(avroSchema); >> >> private static Path testPath; >> >> >> >> >> >> public ParquetTestCase() { >> >> super(TestExecutionMode.COLLECTION); >> >> } >> >> >> >> >> >> @BeforeClass >> >> public static void setup() throws Exception { >> >> >> >> GenericRecordBuilder genericRecordBuilder = new >> GenericRecordBuilder(schm); >> >> >> >> >> >> List recs = new ArrayList<>(); >> >> for (int i = 0; i < 6; i++) { >> >> GenericRecord gr = genericRecordBuilder.set("timestamp_edr", >> System.currentTimeMillis() / 1000).set("id", 333L).set("recordType_", >> "Type1").build(); >> >> recs.add(gr); >> >> GenericRecord gr2 = genericRecordBuilder.set("timestamp_edr", >> System.currentTimeMillis() / 1000).set("id", 22L).set("recordType_", >> "Type2").build(); >> >> recs.add(gr2); >> >> } >> >> >> >> testPath = new Path("/tmp", UUID.randomUUID().toString()); >> >> >> >> >> >> ParquetWriter writer = >> AvroParquetWriter.builder( >> >> new >> org.apache.hadoop.fs.Path(testPath.toUri())).withSchema(schm).build(); >> >> >> >> for (IndexedRecord record : recs) { >> >> writer.write(record); >> >> } >> >> writer.close(); >> >> } >> >> >> >> >> >> private ParquetTableSource createParquetTableSource(Path path) throws >> IOException { >> >> MessageType nestedSchema = SCHEMA_CONVERTER.convert(schm); >> >> ParquetTableSource parquetTableSource = >> ParquetTableSource.builder() >> >> .path(path.getPath()) >> >> .forParquetSchema(nestedSchema) >> >> .build(); >> >> return parquetTableSource; >> >> } >> >> >> >> @Test >> >> public void testScan() throws Exception { >> >> ExecutionEnvironment env = >> ExecutionEnvironment.getExecutionEnvironment(); >> >> >> >> BatchTableEnvironment batchTableEnvironment = >> BatchTableEnvironment.create(env); >> >> ParquetTableSource tableSource = >> createParquetTableSource(testPath); >> >> batchTableEnvironment.registerTableSource("ParquetTable", >> tableSource); >> >> >> >> Table tab = batchTableEnvironment.sqlQuery("select >> id,recordType_ from ParquetTable where id > 22 "); >> >> >> >> DataSet result = batchTableEnvironment.toDataSet(tab, >> Row.class); >> >> >> >> result.print(); >> >> >> >> } >> >> >> >> >> >> } >> >> >> >> >> >> *From:* Peter Huang >> *Sent:* Monday, November 18, 2019 7:22 PM >> *To:* dev >> *Cc:* user@flink.apache.org >> *Subject:* Re: SQL for Avro GenericRecords on Parquet >> >> >> >> Hi Hanan, >> >> >> >> Thanks for reporting the issue. Would you please attach your test code >> here? I may help to investigate. >> >> >> >> >> >> >> >> Best Regards >> >> Peter Huang >> >> >> >> On Mon, Nov 18, 2019 at 2:51 AM Hanan Yehudai >> wrote: >> >> I have tried to persist Generic Avro records in a parquet file and then >> read it via ParquetTablesource – using SQL. >> Seems that the SQL I not executed properly ! >> >> The persisted records are : >> Id , type >> 333,Type1 >> 22,Type2 >> 333,Type1 >> 22,Type2 >> 333,Type1 >> 22,Type2 >> 333,Type1 >> 22,Type2 >> 333,Type1 >> 22,Type2 >> 333,Type1 >> 22,Type2 >> >> While SQL of SELECT id ,recordType_ FROM ParquetTable - return the >> above ( which is correct) >> Running : "SELECT id ,recordType_ FROM ParquetTable where >> recordType_='Type1' " >> Will result in : >> 333,Type1 >> 22,Type1 >> 333,Type1 >> 22,Type1 >> 333,Type1 >> 22,Type1 >> 333,Type1 >> 22,Type1 >> 333,Type1 >> 22,Type1 >> 333,Type1 >> 22,Type1 >> >> As if the equal sign is assignment and not equal … >> >> am I doing something wrong ? is it an issue of Generic record vs >> SpecificRecords ? >> >>