r9p3kpqy

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