· 8 years ago · May 26, 2018, 06:32 PM
1Skip to content
2Features
3Business
4Explore
5Marketplace
6Pricing
7This repository
8Search
9Sign in or Sign up
1035 33 12 Parsely/pyspark-cassandra
11 Code Issues 2 Pull requests 0 Projects 0 Insights
12pyspark-cassandra/src/main/python/pyspark_cassandra_example.py
1358237e8 on 29 Oct 2014
14@msukmanowsky msukmanowsky Initial commit
15
16134 lines (109 sloc) 4.46 KB
17"""
18WARNING: This example is a work in progress!
19Usage:
20 pyspark_cassandra_example.py (init|run) <keyspace> <table>
21Arguments:
22 <command> One of "init" or "run"
23 <keyspace> The name of the keyspace where pixel data is stored
24 <table> The name of the table where pixel data is stored
25"""
26import datetime as dt
27import sys
28from uuid import uuid4
29
30from pyspark.context import SparkConf
31from pyspark_cassandra import CassandraSparkContext, saveToCassandra
32
33
34def create_schemas(keyspace, table):
35 """Utility function to create schemas and tables for example.
36 Requires taht you have the Python Cassandra driver installed on your
37 PYTHONPATH http://datastax.github.io/python-driver/installation.html.
38 """
39 from cassandra.cluster import Cluster
40
41 cluster = Cluster()
42 session = cluster.connect(keyspace)
43
44 # Check to see if the schema/keyspace already exists
45 rows = session.execute("SELECT keyspace_name FROM system.schema_keyspaces;")
46 if not any((row.keyspace_name == keyspace for row in rows)):
47 session.execute("""
48 CREATE SCHEMA {}
49 WITH REPLICATION={'class': 'SimpleStrategy', 'replication_factor': 1};
50 """.format(keyspace))
51 print "Created keyspace: {!r}".format(keyspace)
52 else:
53 print "Keyspace {!r} exists, skipping creation.".format(keyspace)
54
55 session.execute("""
56 CREATE TABLE IF NOT EXISTS {}.{} (
57 customer_id text,
58 url text,
59 hour timestamp,
60 ts timestamp,
61 pixel_id text,
62 data map<text, text>,
63 PRIMARY KEY ((customer_id, url, hour), ts, pixel_id)
64 );
65 """.format(keyspace, table))
66 print "Created table: {!r}.{!r}".format(keyspace, table)
67
68 stmt = session.prepare("""
69 UPDATE {}.{} SET data=? WHERE customer_id=? AND url=? AND hour=?
70 AND ts=? AND pixel_id=?;
71 """.strip().format(keyspace, table))
72
73 pixels = (
74 ({"visitor_id": "1234"}, # data
75 "example.com", # customer_id
76 "http://example.com/", # url
77 dt.datetime(2014, 1, 1, 1), # hour
78 dt.datetime(2014, 1, 1, 1, 23, 12), # ts
79 "8620d3a2-8e03-4f03-bf96-d97369a4c3dc"), # pixel_id
80 ({"visitor_id": "1234"}, "example.com", "http://example.com/",
81 dt.datetime(2014, 1, 1, 1), dt.datetime(2014, 1, 1, 1, 23, 22),
82 "9cab5264-d192-4e0e-ab32-84ebc07d7ed9"),
83 ({"visitor_id": "1234"}, "example.com", "http://example.com/",
84 dt.datetime(2014, 1, 1, 1), dt.datetime(2014, 1, 1, 1, 25, 22),
85 "cb6f1a9e-77d6-4868-a336-c0d736d10d84"),
86 ({"visitor_id": "abcd"}, "example.com", "http://example.com/",
87 dt.datetime(2014, 1, 1, 1), dt.datetime(2014, 1, 1, 1, 25, 22),
88 "c82b1655-1408-4072-b53c-7fd923e8a0c8"),
89 )
90 for pixel in pixels:
91 session.execute(stmt.bind(pixel))
92
93 print "Inserted sample data into: {!r}.{!r}".format(keyspace, table)
94
95
96def run_driver(keyspace, table):
97 conf = SparkConf().setAppName("PySpark Cassandra Sample Driver")
98 conf.set("spark.cassandra.connection.host", "127.0.0.1")
99 sc = CassandraSparkContext(conf=conf)
100
101 # Read some data from Cassandra
102 pixels = sc.cassandraTable(keyspace, table)
103 print pixels.first()
104
105 # Count unique visitors, notice that the data returned by Cassandra is
106 # a dict-like, you can access partition, clustering keys as well as
107 # columns by name. CQL collections: lists, sets and maps are converted
108 # to proper Python data types
109 visitors = pixels.map(lambda p: (p["data"]["visitor_id"],))\
110 .distinct()
111 print "Visitors: {:,}".format(visitors.count())
112
113 # Insert some new pixels into the table
114 pixels = (
115 {
116 "customer_id": "example.com",
117 "url": "http://example.com/article1/",
118 "hour": dt.datetime(2014, 1, 2, 1),
119 "ts": dt.datetime(2014, 1, 2, 1, 8, 23),
120 "pixel_id": str(uuid4()),
121 "data": {"visitor_id": "xyz"}
122 },
123 )
124 saveToCassandra(sc.parallelize(pixels), keyspace, table)
125 print "Wrote new pixels to Cassandra {!r}.{!r}".format(keyspace, table)
126
127
128def main():
129 if len(sys.argv) != 4 or sys.argv[1] not in ("init", "run"):
130 sys.stderr.write(__doc__)
131 sys.exit(-1)
132
133 command = sys.argv[1]
134 keyspace = sys.argv[2]
135 table = sys.argv[3]
136
137 if command == "init":
138 create_schemas(keyspace, table)
139 else:
140 run_driver(keyspace, table)
141
142 print "Done."
143
144
145
146if __name__ == '__main__':
147 main()