<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
Step1:  AWS (S3, Redshift, Kinesis) + Databricks Spark = Real-time Smart Meter Analytics
Create S3 Bucket
Step2:  Copy Postgres to S3 via Postgres dump to CSV and s3cmd upload
Step3:  Amazon Redshift
Step4:  create table electricity (
dataid integer not null,
localhour timestamp not null distkey sortkey,
use decimal(30,26),
air1 decimal(30,26),
furnace1 decimal(30,26),
car1 decimal(30,26)
);
create table weather (
localhour timestamp not null distkey sortkey,
latitude decimal(30,26),
longitude decimal(30,26),
temperature decimal(30,26),
city varchar(20)
);
create table metadata (
dataid integer distkey sortkey,
city varchar(20),
state varchar(20)
);
Step7:  Databricks Spark Analysis (see Databricks)
<ASSISTANT_TASK:>
Python Code:
s3 = boto3.client('s3')
s3.list_buckets()
def create_s3_bucket(bucketname):
    Quick method to create bucket with exception handling
    s3 = boto3.resource('s3')
    exists = True
    bucket = s3.Bucket(bucketname)
    try:
        s3.meta.client.head_bucket(Bucket=bucketname)
    except botocore.exceptions.ClientError as e:
        error_code = int(e.response['Error']['Code'])
        if error_code == 404:
            exists = False
    if exists:
        print 'Bucket {} already exists'.format(bucketname)
    else:
        s3.create_bucket(Bucket=bucketname, GrantFullControl='dkelly628')
create_s3_bucket('pecanstreetresearch-2016')
Explanation: AWS (S3, Redshift, Kinesis) + Databricks Spark = Real-time Smart Meter Analytics
Create S3 Bucket
End of explanation
# Note: Used s3cmd tools because awscli tools not working in conda env
# 14m rows or ~ 1.2 GB local unzipped; 10min write to CSV and another 10min to upload to S3
# !s3cmd put ~/Users/Doug/PecanStreet/electricity-03-06-2016.csv s3://pecanstreetresearch-2016/electricity-03-06-2016.csv
# 200k rows ~ 15 MB local unzipped; 30 sec write to CSV and 15 sec upload to S3
# !s3cmd put ~/Users/Doug/PecanStreet/weather-03-06-2016.csv s3://pecanstreetresearch-2016/weather-03-06-2016.csv
Explanation: Copy Postgres to S3 via Postgres dump to CSV and s3cmd upload
End of explanation
# Quick geohashing before uploading to Redshift
weather_df = pd.read_csv('/Users/Doug/PecanStreet/weather_03-06-2016.csv')
weather_df.groupby(['latitude', 'longitude', 'city']).count()
weather_df['city'] = weather_df['Austin' if weather_df.latitude=30.292432 elif '']
weather_df['city'] = 'city'
weather_df.city.unique()
# weather_df['city'][weather_df.latitude==40.027278] = 'Boulder'
weather_df.to_csv('/Users/Doug/PecanStreet/weather-03-07-2016.csv', index=False)
metadata_df = pd.read_csv('/Users/Doug/PecanStreet/dataport-metadata.csv')
metadata_df = metadata_df[['dataid','city', 'state']]
metadata_df.to_csv('/Users/Doug/PecanStreet/metadata.csv', index=False)
# !s3cmd put metadata.csv s3://pecanstreetresearch-2016/metadata/metadata.csv
redshift = boto3.client('redshift')
# redshift.describe_clusters()
# psql -h pecanstreet.czxmxphrw2wv.us-east-1.redshift.amazonaws.com -U dkelly628 -d electricity -p 5439
Explanation: Amazon Redshift: NoSQL Columnar Data Warehouse
Quick data cleanup before ETL
End of explanation
# Complete
COPY electricity 
FROM 's3://pecanstreetresearch-2016/electricity/electricity-03-06-2016.csv'
CREDENTIALS 'aws_access_key_id=AWS_ACCESS_KEY_ID;aws_secret_access_key=AWS_SECRET_ACCESS_KEY'
CSV
IGNOREHEADER 1
dateformat 'auto';
# Complete
COPY weather
FROM 's3://pecanstreetresearch-2016/weather/weather-03-06-2016.csv'
CREDENTIALS 'aws_access_key_id=AWS_ACCESS_KEY_ID;aws_secret_access_key=AWS_SECRET_ACCESS_KEY'
CSV
IGNOREHEADER 1
dateformat 'auto';
# Complete
COPY metadata 
FROM 's3://pecanstreetresearch-2016/metadata/metadata.csv'
CREDENTIALS 'aws_access_key_id=AWS_ACCESS_KEY_ID;aws_secret_access_key=AWS_SECRET_ACCESS_KEY'
CSV 
IGNOREHEADER 1;
# Query for checking error log; invaluable
select query, substring(filename,22,25) as filename,line_number as line, 
substring(colname,0,12) as column, type, position as pos, substring(raw_line,0,30) as line_text,
substring(raw_field_value,0,15) as field_text, 
substring(err_reason,0,45) as reason
from stl_load_errors 
order by query desc
limit 10;
# All table definitions are stored in pg_table_def table; different from Postgres
SELECT DISTINCT tablename
FROM pg_table_def
WHERE schemaname = 'public'
ORDER BY tablename;
# Returns household, time, city, usage by hour, and temperature for all residents in Austin, TX
SELECT e.dataid, e.localhour, m.city, SUM(e.use), w.temperature
FROM electricity AS e
JOIN weather AS w ON e.localhour = w.localhour
JOIN metadata AS m ON e.dataid = m.dataid 
WHERE m.city = 'Austin'
GROUP BY e.dataid, e.localhour, m.city, w.temperature;
# Returns number of participants by city, state
SELECT m.city, m.state, COUNT(e.dataid) AS participants
FROM electricity AS e
JOIN metadata AS m ON e.dataid = m.dataid
GROUP BY m.city, m.state;
# Setup connection to Pecan Street Dataport
try:
conn = psycopg2.connect("dbname='electricity' user='dkelly628' host='pecanstreet.czxmxphrw2wv.us-east-1.redshift.amazonaws.com' port='5439' password='password'")
except:
#     print "Error: Check there aren't any open connections in notebook or pgAdmin"
electricity_df = pd.read_sql("SELECT localhour, SUM(use) AS usage, SUM(air1) AS cooling, SUM(furnace1) AS heating, \
                             SUM(car1) AS electric_vehicle \
                             FROM electricity \
                             WHERE dataid = 7982 AND use > 0  \
                             AND localhour BETWEEN '2013-10-16 00:00:00'::timestamp AND \
                             '2016-02-26 08:00:00'::timestamp \
                             GROUP BY dataid, localhour \
                             ORDER BY localhour", conn)
electricity_df['localhour'] = electricity_df.localhour.apply(pd.to_datetime)
electricity_df.set_index('localhour', inplace=True)
electricity_df.fillna(value=0.0, inplace=True)
electricity_df[['usage','cooling']].plot(figsize=(18,9), title="Pecan Street Household 7982 Hourly Energy Consumption")
sns.despine();
Explanation: create table electricity (
dataid integer not null,
localhour timestamp not null distkey sortkey,
use decimal(30,26),
air1 decimal(30,26),
furnace1 decimal(30,26),
car1 decimal(30,26)
);
create table weather (
localhour timestamp not null distkey sortkey,
latitude decimal(30,26),
longitude decimal(30,26),
temperature decimal(30,26),
city varchar(20)
);
create table metadata (
dataid integer distkey sortkey,
city varchar(20),
state varchar(20)
);
End of explanation
kinesis = boto3.client('kinesis')
kinesis.create_stream(StreamName='PecanStreet', ShardCount=2)
kinesis.list_streams()
firehose = boto3.client('firehose')
# firehose.create_delivery_stream(DeliveryStreamName='pecanstreetfirehose', S3DestinationConfiguration={'RoleARN': '', 'BucketARN': 'pecanstreetresearch-2016'})
firehose.list_delivery_streams()
def kinesis_write(stream, ):
    Method that writes to kinesis stream
    kinesis = boto3.client('kinesis')
    kinesis.put(StreamName=stream, )
def kinesis_read():
    Method to read from kinesis stream
Explanation: Databricks Spark Analysis (see Databricks): Batch analytics on S3, Streaming using Amazon Kinesis Stream
Create Amazon Kinesis Stream for writing streaming data to S3
End of explanation
<END_TASK>