我們正在創建一個數據流管道，我們將從 postgres 讀取數據并將其寫入鑲木地板文件。ParquetIO.Sink 允許您將 GenericRecord 的 PCollection 寫入 Parquet 文件（來自此處https://beam.apache.org/releases/javadoc/2.5.0/org/apache/beam/sdk/io/parquet/ParquetIO。網頁）。但是鑲木地板文件架構并不像我預期的那樣這是我的模式：schema = new org.apache.avro.Schema.Parser().parse("{\n" +         "     \"type\": \"record\",\n" +         "     \"namespace\": \"com.example\",\n" +         "     \"name\": \"Patterns\",\n" +         "     \"fields\": [\n" +         "       { \"name\": \"id\", \"type\": \"string\" },\n" +         "       { \"name\": \"name\", \"type\": \"string\" },\n" +         "       { \"name\": \"createdAt\", \"type\": {\"type\":\"string\",\"logicalType\":\"timestamps-millis\"} },\n" +         "       { \"name\": \"updatedAt\", \"type\": {\"type\":\"string\",\"logicalType\":\"timestamps-millis\"} },\n" +         "       { \"name\": \"steps\", \"type\": [\"null\",{\"type\":\"array\",\"items\":{\"type\":\"string\",\"name\":\"json\"}}] },\n" +         "     ]\n" +         "}");到目前為止，這是我的代碼：Pipeline p = Pipeline.create(        PipelineOptionsFactory.fromArgs(args).withValidation().create());p.apply(JdbcIO.<GenericRecord> read()       .withDataSourceConfiguration(JdbcIO.DataSourceConfiguration.create(             "org.postgresql.Driver", "jdbc:postgresql://localhost:port/database")             .withUsername("username")             .withPassword("password"))       .withQuery("select * from table limit(10)")       .withCoder(AvroCoder.of(schema))       .withRowMapper((JdbcIO.RowMapper<GenericRecord>) resultSet -> {            GenericRecord record = new GenericData.Record(schema);            ResultSetMetaData metadata = resultSet.getMetaData();            int columnsNumber = metadata.getColumnCount();