from google.cloud.proto.datastore.v1 import entity_pb2
from google.cloud.proto.datastore.v1 import query_pb2
import googledatastore
import apache_beam as beam
from datetime import datetime
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.io.gcp.datastore.v1.datastoreio import ReadFromDatastore
from apache_beam.io.gcp.datastore.v1.datastoreio import WriteToDatastore

project = '[GCLOUD_PROJECT]' # Replace [GCLOUD_PROJECT] with project
kind = 'President'

options = PipelineOptions(project=project)
p = beam.Pipeline(options=options)
lines = p | 'Read from Cloud Storage' >> beam.io.ReadFromText('gs://[GCLOUD_BUCKET]/usa_presidents.csv') # Replace [GCLOUD_BUCKET] with Cloud Storage bucket

def to_entity(line):
  entity = entity_pb2.Entity()
  fields = line.split(',') #id,president,startYear,endYear,party,homeState,dateOfBirth
  id = fields[0]
  president = fields[1]
  names = president.split(' ')
  firstName = names[0]
  lastName = names[1]
  startYear = fields[2]
  endYear = fields[3]
  party = fields[4]
  homeState = fields[5]
  dateOfBirth = fields[6]
  googledatastore.helper.add_key_path(
             entity.key, kind, str(id))
  googledatastore.helper.add_properties(entity, {  
    'firstName': unicode(firstName), 
    'lastName': unicode(lastName), 
    'startYear': int(startYear), 
    'endYear': int(endYear), 
    'party': unicode(party), 
    'homeState': unicode(homeState), 
    'dateOfBirth': datetime.strptime(dateOfBirth, '%Y-%m-%d') })
  return entity


entities = lines | 'To Entity' >> beam.Map(to_entity)
entities | 'Write To Datastore' >> WriteToDatastore(project)
# lines | 'Write to Cloud Storage' >> beam.io.WriteToText('gs://[GCLOUD_BUCKET]/out')

p.run().wait_until_finish()
@@ -1,127 +1,176 @@
-from google.cloud.proto.datastore.v1 import entity_pb2%0Afrom google.cloud.proto.datastore.v1 import query_pb2%0Aimport googled
+import apache_beam as beam%0A%0Afrom apache_beam.io.gcp.datastore.v1new.datastoreio import ReadFromDatastore%0Afrom apache_beam.io.gcp.datastore.v1new.datastoreio import WriteToD
 atas
@@ -174,22 +174,20 @@
 tastore%0A
-import
+from
  apache_
@@ -194,257 +194,329 @@
 beam
- as beam%0Afrom datetime import datetime%0Afrom apache_beam.options.pipeline_options import PipelineOptions%0Afrom apache_beam.io.gcp.datastore.v1.datastoreio import ReadFromDatastore%0Afrom apache_beam.io.gcp.datastore.v1.datastoreio import WriteToDatastor
+.io.gcp.datastore.v1new.types import Entity%0Afrom apache_beam.io.gcp.datastore.v1new.types import Key%0Afrom apache_beam.io.gcp.datastore.v1new.types import Query%0Afrom apache_beam.options.pipeline_options import GoogleCloudOptions%0Afrom apache_beam.options.pipeline_options import PipelineOptions%0Afrom datetime import datetim
 e%0A%0Ap
@@ -602,17 +602,16 @@
 sident'%0A
-%0A
 options 
@@ -645,16 +645,17 @@
 roject)%0A
+%0A
 p = beam
@@ -838,17 +838,16 @@
  bucket%0A
-%0A
 def to_e
@@ -863,39 +863,8 @@
 e):%0A
-  entity = entity_pb2.Entity()%0A
   fi
@@ -953,16 +953,20 @@
 %0A  id = 
+int(
 fields%5B0
@@ -966,16 +966,63 @@
 ields%5B0%5D
+)%0A  key = Key(%5Bkind,id%5D)%0A  entity = Entity(key)
 %0A  presi
@@ -1075,306 +1075,117 @@
 )%0A  
-firstName = names%5B0%5D%0A  lastName = names%5B1%5D%0A  startYear = fields%5B2%5D%0A  endYear = fields%5B3%5D%0A  party = fields%5B4%5D%0A  homeState = fields%5B5%5D%0A  dateOfBirth = fields%5B6%5D%0A  googledatastore.helper.add_key_path(%0A             entity.key, kind, str(id))%0A  googledatastore.helper.add_properties(entity, %7B
+entity.set_properties(%7B'id': id,%0A                         'firstName': names%5B0%5D,%0A                   
   
-%0A
     '
-fir
+la
 stNa
@@ -1193,197 +1193,239 @@
 e': 
-unicode(firstName), %0A    'lastName': unicode(lastName), %0A    'startYear': int(startYear), %0A    'endYear': int(endYear), %0A    'party': unicode(party), %0A    'homeState': unicode(homeState), %0A
+names%5B1%5D,%0A                         'startYear': int(fields%5B2%5D),%0A                         'endYear': int(fields%5B3%5D),%0A                         'party': fields%5B4%5D,%0A                         'homeState': fields%5B5%5D,%0A                     
     
@@ -1457,27 +1457,25 @@
 trptime(
-dateOfBirth
+fields%5B6%5D
 , '%25Y-%25m
@@ -1479,16 +1479,40 @@
 -%25m-%25d')
+%0A                       
  %7D)%0A  re
@@ -1523,18 +1523,16 @@
  entity%0A
-%0A%0A
 entities
@@ -1724,17 +1724,16 @@
 %5D/out')%0A
-%0A
 p.run().
@@ -1751,8 +1751,9 @@
 finish()
+%0A

upgrade to python3 fix demo, now also better matches current examples