[Task]: Support more Beam portable schema types as Python types

[ahmedabu98]

What needs to happen?

Beam portable schemas include primitive and more complex types (represented as logical types). Some of these types are supported in the Python SDK:

beam/sdks/python/apache_beam/typehints/schemas.py

Lines 23 to 41 in 99202b2

	Python Schema
	np.int8 <-----> BYTE
	np.int16 <-----> INT16
	np.int32 <-----> INT32
	np.int64 <-----> INT64
	int ------> INT64
	np.float32 <-----> FLOAT
	np.float64 <-----> DOUBLE
	float ------> DOUBLE
	bool <-----> BOOLEAN
	str <-----> STRING
	bytes <-----> BYTES
	ByteString ------> BYTES
	Timestamp <-----> LogicalType(urn="beam:logical_type:micros_instant:v1")
	Decimal <-----> LogicalType(urn="beam:logical_type:fixed_decimal:v1")
	Mapping <-----> MapType
	Sequence <-----> ArrayType
	NamedTuple <-----> RowType
	beam.Row ------> RowType

When necessary, Python classes are created to represent a portable type. For example, see Timestamp below:

beam/sdks/python/apache_beam/utils/timestamp.py

Line 45 in 99202b2

class Timestamp(object):

There are some missing portable types in the Python SDK (e.g. Date, DateTime, Time) that we should add support for to make the cross-language experience more smooth.

Issue Priority

Priority: 2 (default / most normal work should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[pavleec]

JSON type is also missing in Python SDK 😕

[unography]

Hi @ahmedabu98 , currently GEOGRAPHY as a data type isn't supported, and it throws error here: https://github.com/apache/beam/blob/master/sdks/python/apache_beam/io/gcp/bigquery_tools.py#L110-L123

Are there plans to add support for it?

[jd185367]

I'd also like to bump this as needed for using WriteToBigQuery in Python:

beam/sdks/python/apache_beam/io/gcp/bigquery.py

Line 1887 in fa9eb2f

class WriteToBigQuery(PTransform):

Google recommends using the STORAGE_WRITE_API method in their Dataflow Best Practices, which requires passing this transform the schema argument for a table. But since many of our BigQuery tables have a DATE or DATETIME column, which isn't supported yet for these schemas in Python, we aren't able to use this.

As of Beam 2.60.0, we haven't found a current workaround - e.g. specifying our DATE columns as TIMESTAMP in the Python schema seems to fail either when Beam tries to actually write to BigQuery, or at some point when the Java code is executing and doing its own conversion. If anyone knows a workaround for this, I'd appreciate it.

As a side-note: why does STORAGE_WRITE_API require specifying a schema in advance, while STREAMING_INSERT does not?

[ng-oliver]

hello - i'd also like to bump this

whenever the schema argument provided contains timestamp, date, datetime` the SDK will assume the data has the micro attribute which isn't the case in python native datetime

beam/sdks/python/apache_beam/utils/timestamp.py

Line 63 in 99202b2

self.micros = int(seconds * 1000000) + int(micros)

The legacy method=STREAMING_INSERT is 10x easier to use as it handles reading JSON and writing to BigQuery seamlessly. Appreciate an upgrade on method=STORAGE_API to align with Google's pricing incentive that stroage_api has 2TB free ingestion per month and is half price from streaming_insert on a pay-as-you-go basis.