FilesExpand file tree

 main

/

batch_sample.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

90 lines (70 loc) · 2.81 KB

 main

/

batch_sample.py

Copy path

Top

File metadata and controls

• Code
• Blame

90 lines (70 loc) · 2.81 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

# Copyright 2018 Google Inc. All Rights Reserved.

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

# http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

"""This application demonstrates how to do batch operations using Cloud

Spanner.

For more information, see the README.rst under /spanner.

"""

import argparse

import concurrent.futures

import time

from google.cloud import spanner

# [START spanner_batch_client]

def run_batch_query(instance_id, database_id):

"""Runs an example batch query."""

# Expected Table Format:

# CREATE TABLE Singers (

# SingerId INT64 NOT NULL,

# FirstName STRING(1024),

# LastName STRING(1024),

# SingerInfo BYTES(MAX),

# ) PRIMARY KEY (SingerId);

spanner_client = spanner.Client()

instance = spanner_client.instance(instance_id)

database = instance.database(database_id)

# Create the batch transaction and generate partitions

snapshot = database.batch_snapshot()

partitions = snapshot.generate_read_batches(

table="Singers",

columns=("SingerId", "FirstName", "LastName"),

keyset=spanner.KeySet(all_=True),

)

# Create a pool of workers for the tasks

start = time.time()

with concurrent.futures.ThreadPoolExecutor() as executor:

futures = [executor.submit(process, snapshot, p) for p in partitions]

for future in concurrent.futures.as_completed(futures, timeout=3600):

finish, row_ct = future.result()

elapsed = finish - start

print(u"Completed {} rows in {} seconds".format(row_ct, elapsed))

# Clean up

snapshot.close()

def process(snapshot, partition):

"""Processes the requests of a query in an separate process."""

print("Started processing partition.")

row_ct = 0

for row in snapshot.process_read_batch(partition):

print(u"SingerId: {}, AlbumId: {}, AlbumTitle: {}".format(*row))

row_ct += 1

return time.time(), row_ct

# [END spanner_batch_client]

if __name__ == "__main__":

parser = argparse.ArgumentParser(

description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter

)

parser.add_argument("instance_id", help="Your Cloud Spanner instance ID.")

parser.add_argument(

"database_id", help="Your Cloud Spanner database ID.", default="example_db"

)

args = parser.parse_args()

run_batch_query(args.instance_id, args.database_id)