Jacob Gur created BEAM-7608:
-------------------------------
Summary: v1new ReadFromDatastore skips entities
Key: BEAM-7608
URL: https://issues.apache.org/jira/browse/BEAM-7608
Project: Beam
Issue Type: Bug
Components: io-python-gcp
Affects Versions: 2.13.0
Environment: MacOS 10.14.5, Python 2.7
Reporter: Jacob Gur
A simple map over a datastore kind in local emulator using the new
v1new.datastoreio.ReadFromDatastore skip entities.
The kind has 1516 entities, and when I map over it using the old
ReadFromDatastore transform, it maps all of them, i.e., I can map to id and
write to text file.
But the new transform only maps 365 entities. There is no error. The tail of
the standard output is:
```
INFO:root:Latest stats timestamp for kind face_apilog is 2019-06-18
08:15:21+00:00
INFO:root:Estimated size bytes for query: 116188
INFO:root:Splitting the query into 12 splits
INFO:root:Running
(((GetEntities/Reshuffle/ReshufflePerKey/GroupByKey/Read)+(ref_AppliedPTransform_GetEntities/Reshuffle/ReshufflePerKey/FlatMap(restore_timestamps)_14))+((ref_AppliedPTransform_GetEntities/Reshuffle/RemoveRandomKeys_15)+(ref_AppliedPTransform_GetEntities/Read_16)))+((ref_AppliedPTransform_MapToId_17)+((ref_AppliedPTransform_WriteToFile/Write/WriteImpl/WriteBundles_24)+((ref_AppliedPTransform_WriteToFile/Write/WriteImpl/Pair_25)+((ref_AppliedPTransform_WriteToFile/Write/WriteImpl/WindowInto(WindowIntoFn)_26)+(WriteToFile/Write/WriteImpl/GroupByKey/Write)))))
INFO:root:Running
(WriteToFile/Write/WriteImpl/GroupByKey/Read)+((ref_AppliedPTransform_WriteToFile/Write/WriteImpl/Extract_31)+(ref_PCollection_PCollection_20/Write))
INFO:root:Running
(ref_PCollection_PCollection_12/Read)+((ref_AppliedPTransform_WriteToFile/Write/WriteImpl/PreFinalize_32)+(ref_PCollection_PCollection_21/Write))
INFO:root:Running
(ref_PCollection_PCollection_12/Read)+(ref_AppliedPTransform_WriteToFile/Write/WriteImpl/FinalizeWrite_33)
INFO:root:Starting finalize_write threads with num_shards: 1 (skipped: 0),
batches: 1, num_threads: 1
INFO:root:Renamed 1 shards in 0.12 seconds.
```
The code for the job on the new transform is:
```
from __future__ import absolute_import
import logging
import os
import sys
import apache_beam as beam
from apache_beam.io.gcp.datastore.v1new.datastoreio import ReadFromDatastore
from apache_beam.io.gcp.datastore.v1new.types import Query
# TODO: should be set outside of python process
os.environ['DATASTORE_EMULATOR_HOST'] = 'localhost:8085'
def map_to_id(element):
face_log_id = element.to_client_entity().id
return face_log_id
def run(argv=None):
p = beam.Pipeline(argv=argv)
project = 'fareclockdev'
(p
| 'GetEntities' >> ReadFromDatastore(Query(kind='face_apilog',
project=project))
| 'MapToId' >> beam.Map(map_to_id)
| 'WriteToFile' >> beam.io.WriteToText('result')
)
p.run().wait_until_finish()
if __name__ == '__main__':
logging.getLogger().setLevel(logging.INFO)
run(sys.argv)
```
For comparison, the code for the job on the old transform is:
```
from __future__ import absolute_import
import logging
import os
import sys
import apache_beam as beam
from apache_beam.io.gcp.datastore.v1.datastoreio import ReadFromDatastore
from google.cloud.proto.datastore.v1 import query_pb2
# TODO: should be set outside of python process
os.environ['DATASTORE_EMULATOR_HOST'] = 'localhost:8085'
def map_to_id(element):
face_log_id = element.key.path[-1].id
return face_log_id
def run(argv=None):
p = beam.Pipeline(argv=argv)
project = 'fareclockdev'
query = query_pb2.Query()
query.kind.add().name = 'face_apilog'
(p
| 'GetEntities' >> ReadFromDatastore(project=project, query=query)
# TODO: ParDo???
| 'MapToId' >> beam.Map(map_to_id)
| 'WriteToFile' >> beam.io.WriteToText('result')
)
p.run().wait_until_finish()
if __name__ == '__main__':
logging.getLogger().setLevel(logging.INFO)
run(sys.argv)
```
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
