MallikarjunaG

MallikarjunaG

'''
DataFrames From Complex Schema

@author: Mallikarjuna G
'''

if __name__ == '__main__':
    from pyspark.sql import SparkSession
    from pyspark.sql import functions as F
    from pyspark.sql import column

MallikarjunaG

1:  import sys  
2:  import json  
3:  from pyspark import SparkContext  
4:  from pyspark.streaming import StreamingContext  
5:    
6:    
7:  def SaveRecord(rdd):  
8:    host = 'sparkmaster.example.com'  
9:    table = 'cats'  
10:    keyConv = "org.apache.spark.examples.pythonconverters.StringToImmutableBytesWritableConverter"  

MallikarjunaG

Introduction

This document describes sample process of implementing part of existing Dim_Instance ETL.

I took only Clound Block Storage source to simplify and speedup the process. I also ignnored creation of extended tables (specific for this particular ETL process). Below are code and final thoughts about possible Spark usage as primary ETL tool.

TL;DR

Implementation

Basic ETL implementation is really straightforward. The only real problem (I mean, really problem) is to find correct and comprehensive Mapping document (description what source fields go where).

MallikarjunaG

#!/usr/bin/env python

import sys, os, re
import json
import datetime, iso8601

from pyspark import SparkContext, SparkConf
from pyspark.sql import SparkSession, Row
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils, OffsetRange, TopicAndPartition

MallikarjunaG

// Use Gists to store code you would like to remember later on
console.log(window); // log the "window" object to the console