Framed Data

Startup the Spark shell, and load the data file, into an RDD[String]:

$ spark-shell

var tx=sc.textFile("file:///home/dmn/city_data/cities1000.txt")  

tx.count()
Long = 145725

Define a case class, and a parse function :

case class City(
        geonameid: Int, 
        name: String, 
        asciiname: String, 
        latitude: Double, longitude: Double, 
        country: String, 
        population: Int, 
        elevation: Int) 

def parse(line: String) = { 
  val spl=line.split("\t") 
  val geonameid=spl(0).toInt
  val name=spl(1)
  val asciiname=spl(2)
  val latitude=spl(4).toDouble
  val longitude=spl(5).toDouble
  val country=spl(8)
  val population=spl(14).toInt
  val elevation=spl(16).toInt
  City(geonameid, name, asciiname, latitude, longitude, country, population, elevation)
}

Try and parse 1 line:

parse(tx.take(1)(0))
City = City(3039154,El Tarter,El Tarter,42.57952,1.65362,AD,1052,1721)

Success! Now let's parse the complete text file into City records:

var ct=tx.map(parse(_))

Check:

ct.count
Long = 145725

Spot-check: list all cities above 3500m and having a population of more than 100000, ordered by descending elevation:

var chk=ct.filter( rec => ( rec.elevation>3500) && (rec.population>100000)).collect()
chk.sortWith( (x,y) => (x.elevation>y.elevation) ).foreach(println)

City(3907584,Potosí,Potosi,-19.58361,-65.75306,BO,141251,3967)
City(3909234,Oruro,Oruro,-17.98333,-67.15,BO,208684,3936)
City(3937513,Juliaca,Juliaca,-15.5,-70.13333,PE,245675,3834)
City(3931276,Puno,Puno,-15.8422,-70.0199,PE,116552,3825)
City(3911925,La Paz,La Paz,-16.5,-68.15,BO,812799,3782)
City(1280737,Lhasa,Lhasa,29.65,91.1,CN,118721,3651)

Startup the Spark shell, and load the data file, into an RDD[String]:

$ spark-shell

var tx=sc.textFile("file:///home/dmn/city_data/cities1000.txt")  

tx.count()
Long = 145725

Import some needed types:

import org.apache.spark.sql.types.{StructType,StructField,StringType,LongType,DoubleType};
import org.apache.spark.sql.Row;

Create the SQL context:

val sqlctx= new org.apache.spark.sql.SQLContext(sc)

Define the schema:

val schema = StructType(Array(
    StructField("geonameid",LongType,false),
    StructField("name",StringType,true),
    StructField("asciiname",StringType,true),
    StructField("latitude",DoubleType,true),
    StructField("longitude",DoubleType,true),
    StructField("country",StringType,true),
    StructField("population",LongType,true),
    StructField("elevation",LongType,true)
    ) )

Turn the text-line RDD into a row RDD:

val rowrdd = tx.map(_.split("\t")).map(p =>
        Row(p(0).toLong,p(1), p(2),
            p(4).toDouble,p(5).toDouble,
            p(8),p(14).toLong,p(16).toLong) )

Create the dataframe:

val city_df=sqlctx.createDataFrame(rowrdd,schema)

Spotcheck:

city_df.filter(" elevation>3500 and population>100000 ").
        orderBy(desc("elevation")).
        collect().
        foreach(println)

[3907584,Potosí,Potosi,-19.58361,-65.75306,BO,141251,3967]
[3909234,Oruro,Oruro,-17.98333,-67.15,BO,208684,3936]
[3937513,Juliaca,Juliaca,-15.5,-70.13333,PE,245675,3834]
[3931276,Puno,Puno,-15.8422,-70.0199,PE,116552,3825]
[3911925,La Paz,La Paz,-16.5,-68.15,BO,812799,3782]
[1280737,Lhasa,Lhasa,29.65,91.1,CN,118721,3651]

See tab Spark Dataframe, on how to create the dataframe city_df :

Then register the table:

city_df.registerTempTable("city")

And Bob's your uncle.

Spotcheck:

sqlctx.sql("select * from city where elevation>3500 and population>100000 order by elevation desc").
       collect().foreach(println)

[3907584,Potosí,Potosi,-19.58361,-65.75306,BO,141251,3967]
[3909234,Oruro,Oruro,-17.98333,-67.15,BO,208684,3936]
[3937513,Juliaca,Juliaca,-15.5,-70.13333,PE,245675,3834]
[3931276,Puno,Puno,-15.8422,-70.0199,PE,116552,3825]
[3911925,La Paz,La Paz,-16.5,-68.15,BO,812799,3782]
[1280737,Lhasa,Lhasa,29.65,91.1,CN,118721,3651]

Note: if you run above without the collect() then the ordering may be incorrect.

df<-read.table("/home/dmn/city_data/cities1000.txt",sep="\t"
                 ,quote="",stringsAsFactors=F, na.strings = "",
                 )[,c(1, 2, 3, 5, 6, 9, 15, 17)]
names(df)<-c("geonameid","name","asciiname","latitude","longitude",
             "country","population","elevation")

options(width=200)

Note: if you don't set na.strings = "" in this case, then all Namibia (code 'NA') records are flagged as having NA for country. Check: table(is.na(df$country)) should give all F's and no T's.

Dimensions:

dim(df)
145725      8

Spotcheck: cities above 3500m and having a population of more than 100000:

tf=df[(df$elevation>3500)&(df$population>100000),]
tf[order(-tf$elevation),] 

       geonameid    name asciiname  latitude longitude country population elevation
7015     3907584  Potosí    Potosi -19.58361 -65.75306      BO     141251      3967
7026     3909234   Oruro     Oruro -17.98333 -67.15000      BO     208684      3936
101370   3937513 Juliaca   Juliaca -15.50000 -70.13333      PE     245675      3834
101162   3931276    Puno      Puno -15.84220 -70.01990      PE     116552      3825
7044     3911925  La Paz    La Paz -16.50000 -68.15000      BO     812799      3782
12297    1280737   Lhasa     Lhasa  29.65000  91.10000      CN     118721      3651

Read tab-separated file into a data.table

Read:

library(data.table)
dt<-fread("/home/dmn/city_data/cities1000.txt", header=FALSE,
                na.strings = "",
                select=c(1, 2, 3, 5, 6, 9, 15, 17) )
setnames(dt,c("geonameid","name","asciiname","latitude","longitude",
              "country","population","elevation"))

options(width=200)

Spotcheck: cities above 3500m and having a population of more than 100000:

dt[(dt$elevation>3500)&(dt$population>100000),][order(-elevation),]

   geonameid    name asciiname  latitude longitude country population elevation
1:   3907584  Potosí    Potosi -19.58361 -65.75306      BO     141251      3967
2:   3909234   Oruro     Oruro -17.98333 -67.15000      BO     208684      3936
3:   3937513 Juliaca   Juliaca -15.50000 -70.13333      PE     245675      3834
4:   3931276    Puno      Puno -15.84220 -70.01990      PE     116552      3825
5:   3911925  La Paz    La Paz -16.50000 -68.15000      BO     812799      3782
6:   1280737   Lhasa     Lhasa  29.65000  91.10000      CN     118721      3651

import pandas as pd
import csv   

colnames= [ "geonameid","name","asciiname","latitude","longitude",
            "country","population","elevation" ]

df=pd.io.parsers.read_table("/home/dmn/city_data/cities1000.txt",
                sep="\t", header=None, names= colnames,
                quoting=csv.QUOTE_NONE,usecols=[ 0, 1, 2, 4, 5, 8, 14, 16])

pd.set_option('display.width', 200)

Number of records:

len(df)
145725

Spotcheck: cities above 3500m and having a population of more than 100000:

df[(df.elevation>3500) & (df.population>100000)].sort("elevation",ascending=False)

        geonameid     name asciiname  latitude  longitude country  population  elevation
7014      3907584   Potosí    Potosi -19.58361  -65.75306      BO      141251       3967
7025      3909234    Oruro     Oruro -17.98333  -67.15000      BO      208684       3936
100330    3937513  Juliaca   Juliaca -15.50000  -70.13333      PE      245675       3834
100122    3931276     Puno      Puno -15.84220  -70.01990      PE      116552       3825
7043      3911925   La Paz    La Paz -16.50000  -68.15000      BO      812799       3782
12296     1280737    Lhasa     Lhasa  29.65000   91.10000      CN      118721       3651

Preprocessing

The data will be loaded into a Postgres table using the copy command. With this command, it's not possible to only load a subset of columns, therefore we are using R to preprocess the file.

Startup R, read selected columns of the input, then copy the data out into file "cities.csv" :

df<-read.table("/home/dmn/city_data/cities1000.txt", sep="\t",
               quote="",stringsAsFactors=F)[,c(1, 2, 3, 5, 6, 9, 15, 17)]

# corrections for city names with a double quote in their name:
df[df$V1==1682560,c("V2","V3")]=c("T-boli","T-boli")
df[df$V1==688022,c("V2","V3")]=c("Yur-yivka","Yur-yivka")

write.table(df,"cities.csv", sep="\t",row.names=F,col.names=F,na="",quote=F)

Sidenote: create a postgres database user

On the linux CLI :

sudo  -u postgres psql template1
template1=# create user dmn encrypted password 'dmn';
template1=# create DATABASE dmn WITH TEMPLATE = template0 ENCODING = 'UTF8' owner dmn;
template1=# \q

Startup psql :

psql -h localhost -U dmn

Load into postgres

Startup psql, and create the table:

create table t_city (
     geonameid int  primary key
    ,name      varchar(128) 
    ,asciiname varchar(128) 
    ,latitude  numeric 
    ,longitude  numeric
    ,country   varchar(5) 
    ,population int
    ,elevation int
);

Load the data:

 \copy t_city from 'cities.csv' with DELIMITER E'\t' CSV

Count:

select count(1) from t_city 

count  
--------
 145725

Spotcheck: cities above 3500m and having a population of more than 100000:

select * 
from t_city 
where elevation>3500 and population>100000
order by elevation desc;

 geonameid |  name   | asciiname | latitude  | longitude | country | population | elevation 
-----------+---------+-----------+-----------+-----------+---------+------------+-----------
   3907584 | Potosí  | Potosi    | -19.58361 | -65.75306 | BO      |     141251 |      3967
   3909234 | Oruro   | Oruro     | -17.98333 |    -67.15 | BO      |     208684 |      3936
   3937513 | Juliaca | Juliaca   |     -15.5 | -70.13333 | PE      |     245675 |      3834
   3931276 | Puno    | Puno      |  -15.8422 |  -70.0199 | PE      |     116552 |      3825
   3911925 | La Paz  | La Paz    |     -16.5 |    -68.15 | BO      |     812799 |      3782
   1280737 | Lhasa   | Lhasa     |     29.65 |      91.1 | CN      |     118721 |      3651

Load the data from a csv file

Read tab-separated file into a data.table

Preprocessing

Sidenote: create a postgres database user

Load into postgres