This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new e0f4a0ca927 [FLINK-28808][csv] Create mapper on server-side
e0f4a0ca927 is described below
commit e0f4a0ca9271f0e9e31011578dbdea4b5a8d30fe
Author: Chesnay Schepler <[email protected]>
AuthorDate: Fri Aug 5 19:16:57 2022 +0200
[FLINK-28808][csv] Create mapper on server-side
---
.../flink/formats/csv/CsvFileFormatFactory.java | 27 ++++++++++++----------
.../flink/formats/csv/RowDataToCsvConverters.java | 2 +-
flink-python/pom.xml | 6 +++++
flink-python/pyflink/datastream/formats/csv.py | 12 ++--------
.../apache/flink/formats/csv/PythonCsvUtils.java | 10 ++++----
5 files changed, 30 insertions(+), 27 deletions(-)
diff --git
a/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFileFormatFactory.java
b/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFileFormatFactory.java
index 391e1c49f20..b0a0b27d5e7 100644
---
a/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFileFormatFactory.java
+++
b/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFileFormatFactory.java
@@ -163,18 +163,7 @@ public class CsvFileFormatFactory implements
BulkReaderFormatFactory, BulkWriter
final RowType rowType = (RowType)
physicalDataType.getLogicalType();
final CsvSchema schema = buildCsvSchema(rowType,
formatOptions);
- final RowDataToCsvConverter converter =
- RowDataToCsvConverters.createRowConverter(rowType);
-
- final CsvMapper mapper = new CsvMapper();
- final ObjectNode container = mapper.createObjectNode();
-
- final RowDataToCsvConverter.RowDataToCsvFormatConverterContext
converterContext =
- new
RowDataToCsvConverter.RowDataToCsvFormatConverterContext(
- mapper, container);
-
- return out ->
- CsvBulkWriter.forSchema(mapper, schema, converter,
converterContext, out);
+ return createCsvBulkWriterFactory(schema, rowType);
}
@Override
@@ -184,6 +173,20 @@ public class CsvFileFormatFactory implements
BulkReaderFormatFactory, BulkWriter
};
}
+ static BulkWriter.Factory<RowData> createCsvBulkWriterFactory(
+ CsvSchema schema, RowType rowType) {
+ final RowDataToCsvConverter converter =
RowDataToCsvConverters.createRowConverter(rowType);
+
+ return out -> {
+ final CsvMapper mapper = new CsvMapper();
+ final ObjectNode container = mapper.createObjectNode();
+
+ final RowDataToCsvConverter.RowDataToCsvFormatConverterContext
converterContext =
+ new
RowDataToCsvConverter.RowDataToCsvFormatConverterContext(mapper, container);
+ return CsvBulkWriter.forSchema(mapper, schema, converter,
converterContext, out);
+ };
+ }
+
private static CsvSchema buildCsvSchema(RowType rowType, ReadableConfig
options) {
final CsvSchema csvSchema = CsvRowSchemaConverter.convert(rowType);
final CsvSchema.Builder csvBuilder = csvSchema.rebuild();
diff --git
a/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/RowDataToCsvConverters.java
b/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/RowDataToCsvConverters.java
index 693e89198e2..8d689086558 100644
---
a/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/RowDataToCsvConverters.java
+++
b/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/RowDataToCsvConverters.java
@@ -65,7 +65,7 @@ public class RowDataToCsvConverters implements Serializable {
* Converter context for passing the {@code CsvMapper} and the {@code
container} that can be
* reused between transformations of the individual elements for
performance reasons.
*/
- class RowDataToCsvFormatConverterContext implements Serializable {
+ class RowDataToCsvFormatConverterContext {
CsvMapper csvMapper;
ContainerNode<?> container;
diff --git a/flink-python/pom.xml b/flink-python/pom.xml
index 2ee84a9078d..6d5b18385b3 100644
--- a/flink-python/pom.xml
+++ b/flink-python/pom.xml
@@ -77,6 +77,12 @@ under the License.
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-files</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-csv</artifactId>
diff --git a/flink-python/pyflink/datastream/formats/csv.py
b/flink-python/pyflink/datastream/formats/csv.py
index e475684035c..e5d18dd2309 100644
--- a/flink-python/pyflink/datastream/formats/csv.py
+++ b/flink-python/pyflink/datastream/formats/csv.py
@@ -340,17 +340,9 @@ class CsvBulkWriter(object):
Builds a :class:`BulkWriterFactory` for writing records to files in
CSV format.
"""
jvm = get_gateway().jvm
- jackson = jvm.org.apache.flink.shaded.jackson2.com.fasterxml.jackson
csv = jvm.org.apache.flink.formats.csv
- j_converter = csv.RowDataToCsvConverters.createRowConverter(
- _to_java_data_type(schema._row_type).getLogicalType())
- j_mapper = jackson.dataformat.csv.CsvMapper()
- j_container = j_mapper.createObjectNode()
- j_context =
csv.PythonCsvUtils.createRowDataToCsvFormatConverterContext(
- j_mapper, j_container)
-
j_factory = csv.PythonCsvUtils.createCsvBulkWriterFactory(
- j_mapper, schema._j_schema, j_converter, j_context
- )
+ schema._j_schema,
+ _to_java_data_type(schema._row_type))
return RowDataBulkWriterFactory(j_factory, schema._row_type)
diff --git
a/flink-python/src/main/java/org/apache/flink/formats/csv/PythonCsvUtils.java
b/flink-python/src/main/java/org/apache/flink/formats/csv/PythonCsvUtils.java
index 8202d46c800..08c08a80318 100644
---
a/flink-python/src/main/java/org/apache/flink/formats/csv/PythonCsvUtils.java
+++
b/flink-python/src/main/java/org/apache/flink/formats/csv/PythonCsvUtils.java
@@ -18,7 +18,7 @@
package org.apache.flink.formats.csv;
import org.apache.flink.api.common.serialization.BulkWriter;
-import org.apache.flink.formats.common.Converter;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.RowType;
@@ -62,8 +62,10 @@ public class PythonCsvUtils {
/**
* Util for creating a {@link BulkWriter.Factory} that wraps {@link
CsvBulkWriter#forSchema}.
*/
- public static <T, R, C> BulkWriter.Factory<T> createCsvBulkWriterFactory(
- CsvMapper mapper, CsvSchema schema, Converter<T, R, C> converter,
C converterContext) {
- return (out) -> CsvBulkWriter.forSchema(mapper, schema, converter,
converterContext, out);
+ public static BulkWriter.Factory<RowData> createCsvBulkWriterFactory(
+ CsvSchema schema, DataType physicalDataType) {
+
+ return CsvFileFormatFactory.createCsvBulkWriterFactory(
+ schema, (RowType) physicalDataType.getLogicalType());
}
}
