Big Data | Learn for Master

run pyspark on oozie

July 29, 2016 Author: david

In this post, I first give a workable example to run pySpark on oozie. Then I show how to run pyspark on oozie using your own python installation (e.g., anaconda). In this way, you can use numpy, pandas, other python libraries in your pyspark program.

The syntax of creating a spark action on oozie workflow

<workflow-app name="[WF-DEF-NAME]" xmlns="uri:oozie:workflow:0.3">
    ...
    <action name="[NODE-NAME]">
        <spark xmlns="uri:oozie:spark-action:0.1">
            <job-tracker>[JOB-TRACKER]</job-tracker>
            <name-node>[NAME-NODE]</name-node>
            <prepare>
               <delete path="[PATH]"/>
               ...
               <mkdir path="[PATH]"/>
               ...
            </prepare>
            <job-xml>[SPARK SETTINGS FILE]</job-xml>
            <configuration>
                <property>
                    <name>[PROPERTY-NAME]</name>
                    <value>[PROPERTY-VALUE]</value>
                </property>
                ...
            </configuration>
            <master>[SPARK MASTER URL]</master>
            <mode>[SPARK MODE]</mode>
            <name>[SPARK JOB NAME]</name>
            <class>[SPARK MAIN CLASS]</class>
            <jar>[SPARK DEPENDENCIES JAR / PYTHON FILE]</jar>
            <spark-opts>[SPARK-OPTIONS]</spark-opts>
            <arg>[ARG-VALUE]</arg>
                ...
            <arg>[ARG-VALUE]</arg>
            ...
        </spark>
        <ok to="[NODE-NAME]"/>
        <error to="[NODE-NAME]"/>
    </action>
    ...
</workflow-app>

<workflow-app name="[WF-DEF-NAME]" xmlns="uri:oozie:workflow:0.3">

...

<job-tracker>[JOB-TRACKER]</job-tracker>

<name-node>[NAME-NODE]</name-node>

...

...

</prepare>

<job-xml>[SPARK SETTINGS FILE]</job-xml>

<name>[PROPERTY-NAME]</name>

<value>[PROPERTY-VALUE]</value>

</property>

...

</configuration>

<master>[SPARK MASTER URL]</master>

<mode>[SPARK MODE]</mode>

<name>[SPARK JOB NAME]</name>

<class>[SPARK MAIN CLASS]</class>

<jar>[SPARK DEPENDENCIES JAR / PYTHON FILE]</jar>

<spark-opts>[SPARK-OPTIONS]</spark-opts>

<arg>[ARG-VALUE]</arg>

...

<arg>[ARG-VALUE]</arg>

...

</spark>

</action>

...

</workflow-app>

As described in the document, here are the meanings of these elements.

The prepare element, if present, indicates a list of paths to delete or create before starting the job. Specified paths must start with hdfs://HOST:PORT .

[Read More...]

Parameter Server 资料汇总

July 6, 2016 Author: david
No comment yet

parameter server 介绍
作者：Superjom
链接：https://www.zhihu.com/question/26998075/answer/40577680
来源：知乎
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。
看看李沐的文章《Parameter Server for Distributed Machine Learning》里面有包含他的框架的一些介绍。
后面有看到微软研究院 project Adam的论文，大体思路比较相似，但论文中细节比较丰富，也会互补的一些信息描述下。

概念：
参数服务器是个编程框架，用于方便分布式并行程序的编写，其中重点是对大规模参数的分布式存储和协同的支持。

工业界需要训练大型的机器学习模型，一些广泛使用的特定的模型在规模上的两个特点：
1. 参数很大，超过单个机器的容纳能力（比如大型Logistic Regression和神经网络）
2. 训练数据巨大，需要分布式并行提速（大数据）

这种需求下，当前类似MapReduce的框架并不能很好适合。
因此需要自己实现分布式并行程序，其实在Hadoop出来之前，对于大规模数据的处理，都需要自己写分布式的程序（MPI）。之后这方面的工作流程被Google的工程师总结和抽象成MapReduce框架，大一统了。

参数服务器就类似于MapReduce，是大规模机器学习在不断使用过程中，抽象出来的框架之一。重点支持的就是参数的分布式，毕竟巨大的模型其实就是巨大的参数。

Parameter Server(Mli)
—————————-
架构：
集群中的节点可以分为计算节点和参数服务节点两种。其中，计算节点负责对分配到自己本地的训练数据（块）计算学习，并更新对应的参数；参数服务节点采用分布式存储的方式，各自存储全局参数的一部分，并作为服务方接受计算节点的参数查询和更新请求。

简而言之吧，计算节点负责干活和更新参数，参数服务节点则负责存储参数。

冗余和恢复：
类似MapReduce，每个参数在参数服务器的集群中都在多个不同节点上备份（3个也是极好的），这样当出现节点失效时，冗余的参数依旧能够保证服务的有效性。当有新的节点插入时，把原先失效节点的参数从冗余参数那边复制过来，失效节点的接班人就加入队伍了。

并行计算：
并行计算这部分主要在计算节点上进行。类似于MapReduce，分配任务时，会将数据拆分给每个worker节点。
参数服务器在开始学习前，也会把大规模的训练数据拆分到每个计算节点上。单个计算节点就对本地数据进行学习就可以了。学习完毕再把参数的更新梯度上传给对应的参数服务节点进行更新。

详细的流程：

1.
分发训练数据 -> 节点1
节点2
节点3
…
节点i
…
节点N

2.
[Read More...]
Resources for Learning Big Data

June 30, 2016 Author: david
No comment yet
Big Data is a really hot topic! There are millions of job openings. This post aims to introduce some of the most important techniques related to Big Data. If you want to learn Big Data and find a job related to Big Data, please read the articles and books recommended here.

The first key concept of Big Data is MapReduce, which is the core of Hadoop, a open sourced framework that becomes the foundation of the Big Data eco system.
- MapReduce was invented by Google. It’s a paradigm for writing distributed systems inspired by some elements of functional programming.
[Read More...]

Append to a Hive partition from Pig

June 27, 2016 Author: david

No comment yet

When we use Hive, we can append data to the table easily, but when we use Pig (i.e., the HCatalog ) to insert data into Hive table, we are not allowed to append data to a partition if that partition already contains data.

In this post, I describe a method that can help you append data to the existing partition using a dummy partition named run. It means the run number you append some data to this partition.

For example, we create the following partitioned hive table:

CREATE TABLE table_part (
    id  INT,
    user_id    String,
    url    String
)
PARTITIONED BY (date STRING, run STRING)
row format delimited
fields terminated by ','
stored as textfile;

CREATE TABLE table_part (

id INT,

user_id String,

url String

)

PARTITIONED BY (date STRING, run STRING)

row format delimited

fields terminated by ','

stored as textfile;

Then pig script looks like the following:

-bash-4.1$ cat save_simple.pig
data = load 'hdfs:hostname:port/tmp/user/simple.txt' Using PigStorage(',') as (id:int, user_id:chararray, url:chararray);
--dump data;
store data into 'cb_mappi_db.table_part' using org.apache.hive.hcatalog.pig.HCatStorer('date=${DATE}, run=${RUN}');

-bash-4.1$ cat save_simple.pig

data = load 'hdfs:hostname:port/tmp/user/simple.txt' Using PigStorage(',') as (id:int, user_id:chararray, url:chararray);

--dump data;

store data into 'cb_mappi_db.table_part' using org.apache.hive.hcatalog.pig.HCatStorer('date=${DATE}, run=${RUN}');

Now we can run the pig script using the following command:

-bash-4.1$ cat run_save_simple.sh
pig -useHCatalog -param DATE=20160605 \
-param RUN=1 \
save_simple.pig

-bash-4.1$ cat run_save_simple.sh

pig -useHCatalog -param DATE=20160605 \

-param RUN=1 \

save_simple.pig

Then we have the following content in the table:

hive> select * from table_part;
select * from table_part;
OK
1 u1 url1 20160605 1
2 u2 url2 20160605 1
1 u1 url2 20160605 1
2 u3 url4 20160605 1

hive> select * from table_part;

select * from table_part;

1 u1 url1 20160605 1

2 u2 url2 20160605 1

1 u1 url2 20160605 1

2 u3 url4 20160605 1

Each time when you want to append data to the partition DATE=20160605,

[Read More...]

Set variable for hive script

June 27, 2016 Author: david
1 Comment

When we run hive scripts, such as Load data into Hive table, we often need to pass parameters to the hive scripts by defining our own variables.

Here are some examples to show how to pass parameters or user defined variables to hive.

Use hiveconf for variable subsititution

For example, you can define a variable DATE, then use it as ${hiveconf:DATE}

hive> set DATE='20150405'; hive> select * from mytable where day >= '${hiveconf:DATE}'

1
2
3

hive> set DATE='20150405';
hive> select * from mytable where day >= '${hiveconf:DATE}'

you can even pass the variable from command line:

% hive -hiveconf DATE='20150405' -f my_hive.hql

1
2
3

% hive -hiveconf DATE='20150405' -f my_hive.hql

Use env and system variables

You can also use env and system variables like this ${env:USER}

You can run the following command to see all the available variables:

% hive -e 'set;'

1
2
3

% hive -e 'set;'

If you are o the hive prompt,
[Read More...]
An Example to Create a Partitioned Hive Table

June 25, 2016 Author: david
1 Comment

Partition is a very useful feature of Hive. Without partition, it is hard to reuse the Hive Table if you use HCatalog to store data to Hive table using Apache Pig, as you will get exceptions when you insert data to a non-partitioned Hive Table that is not empty.

In this post, I use an example to show how to create a partitioned table, and populate data into it.

Let’s suppose you have a dataset for user impressions. For instance, a sample of the data set might be like this:

id
user_id
user_lang
user_device
time_stamp
url
date
country

1
u1
en
iphone
201503210011
http://xxx/xxx/1
20150321
US

2
u1
en
ipad
201503220111
http://xxx/xxx/2
20150322
US

3
u2
en
desktop
201503210051
http://xxx/xxx/3
20150321
CA

4
u3
en
iphone
201503230021
http://xxx/xxx/4
20150323
HK

If you use Pig to analyze the data,
[Read More...]
Exceptions When Delete rows from Hive Table

June 25, 2016 Author: david
No comment yet

It’s straight forward to delete data from a traditional Relational table using SQL. However, delete rows from Hive Rows can cause several exceptions.

For examples, let see we have a imps_part table, we want to delete the values in the Table. You will get the exception:

When you run the simple delete command, we get: FAILED: SemanticException [Error 10294]: Attempt to do update or delete using transaction manager that does not support these operations

hive> delete from imps_part;

1

hive> delete from imps_part;

Some one suggest to use the following command:

hive> INSERT OVERWRITE TABLE imps_part > SELECT * FROM imps_part > WHERE id = 1;

1
2
3

hive> INSERT OVERWRITE TABLE imps_part
> SELECT * FROM imps_part
> WHERE id = 1;

This will result in the following exception:
FAILED: SemanticException 1:23 Need to specify partition columns because the destination table is partitioned.
[Read More...]

Save data to Hive table Using Apache Pig

June 24, 2016 Author: david

2 Comments

We have described how to load data from Hive Table using Apache Pig, in this post, I will use an example to show how to save data to Hive table using Pig.

Before save data to Hive, you need to first create a Hive Table. Please refer to this post on how to create a Hive table.

Suppose we use Apache Pig to Load some data from a text file, then we can save the data to the hive table using the following script.

-bash-4.1$ cat student.txt
1 san 19 89
2 LILI 23 90
3 Jim 20 99

-bash-4.1$ cat student.txt

1 san 19 89

2 LILI 23 90

3 Jim 20 99

The store_student.pig script is like this:

data = load 'student.txt' Using PigStorage('\t') as (id:Int, name:chararray, age:Int, score:Float);
STORE data into 'students_db.student_orc' USING org.apache.hive.hcatalog.pig.<code>HCatStorer()</code>;

1 2	data = load 'student.txt' Using PigStorage('\t') as (id:Int, name:chararray, age:Int, score:Float); STORE data into 'students_db.student_orc' USING org.apache.hive.hcatalog.pig.<code>HCatStorer()</code>;

Note: You must specify the table name in single quotes: STORE data into ‘tablename’.

[Read More...]

How to get hive table delimiter or schema

June 24, 2016 Author: david

No comment yet

When you have a hive table, you may want to check its delimiter or detailed information such as Schema. There are two solutions:

1. show create table <table_name>

2. describe extended <table_name>; 

3. describe formatted <table_name>;

1. show create table <table_name>

2. describe extended <table_name>;

3. describe formatted <table_name>;

Get the delimiter of a Hive Table

To get the field delimiter of a hive table, we can use the following command:

show create table <table_name>

1	show create table <table_name>

Here is an example:

hive> show create table student;
OK
CREATE TABLE `student`(
 `id` int,
 `name` string,
 `age` int,
 `score` float)
COMMENT 'Description of the table'
ROW FORMAT DELIMITED
 FIELDS TERMINATED BY '\t'
 LINES TERMINATED BY '\n'
STORED AS INPUTFORMAT
 'org.apache.hadoop.mapred.TextInputFormat'
OUTPUTFORMAT
 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION
 'hdfs://.../tmp/xx/hivedb/student'
TBLPROPERTIES (
 'COLUMN_STATS_ACCURATE'='false',
 'hcat.msgbus.topic.name'='hcat.xxxx.students_db.student',
 'numFiles'='0',
 'numRows'='-1',
 'rawDataSize'='-1',
 'totalSize'='0',
 'transient_lastDdlTime'='1466792036')
Time taken: 0.149 seconds, Fetched: 23 row(s)

hive> show create table student;

CREATE TABLE `student`(

`id` int,

`name` string,

`age` int,

`score` float)

COMMENT 'Description of the table'

ROW FORMAT DELIMITED

FIELDS TERMINATED BY '\t'

LINES TERMINATED BY '\n'

STORED AS INPUTFORMAT

'org.apache.hadoop.mapred.TextInputFormat'

OUTPUTFORMAT

'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'

LOCATION

'hdfs://.../tmp/xx/hivedb/student'

TBLPROPERTIES (

'COLUMN_STATS_ACCURATE'='false',

'hcat.msgbus.topic.name'='hcat.xxxx.students_db.student',

'numFiles'='0',

'numRows'='-1',

'rawDataSize'='-1',

'totalSize'='0',

'transient_lastDdlTime'='1466792036')

Time taken: 0.149 seconds, Fetched: 23 row(s)

Get the schema of Hive Table

Another solution is to use:

describe extended <table_name>;

1	describe extended <table_name>;

This will generate a competed information about the table.

hive> describe extended student;
OK
id int
name string
age int
score float

Detailed Table Information Table(tableName:student, dbName:students_db, owner:xxx, createTime:1466792036, lastAccessTime:0, retention:0, sd:StorageDescriptor(cols:[FieldSchema(name:id, type:int, comment:null), FieldSchema(name:name, type:string, comment:null), FieldSchema(name:age, type:int, comment:null), FieldSchema(name:score, type:float, comment:null)], location:hdfs://xxx-nn1.blue.grid.xxx.com:8020/tmp/xxx/hivedb/student, inputFormat:org.apache.hadoop.mapred.TextInputFormat, outputFormat:org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat, compressed:false, numBuckets:-1, serdeInfo:SerDeInfo(name:null, serializationLib:org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, parameters:{line.delim=
, field.delim= , serialization.format= }), bucketCols:[], sortCols:[], parameters:{}, skewedInfo:SkewedInfo(skewedColNames:[], skewedColValues:[], skewedColValueLocationMaps:{}), storedAsSubDirectories:false), partitionKeys:[], parameters:{hcat.msgbus.topic.name=hcat.jet-blue.students_db.student, totalSize=0, numRows=-1, rawDataSize=-1, COLUMN_STATS_ACCURATE=false, numFiles=0, transient_lastDdlTime=1466792036, comment=Description of the table}, viewOriginalText:null, viewExpandedText:null, tableType:MANAGED_TABLE)
Time taken: 0.048 seconds, Fetched: 7 row(s)

hive> describe extended student;

id int

name string

age int

score float

Detailed Table Information Table(tableName:student, dbName:students_db, owner:xxx, createTime:1466792036, lastAccessTime:0, retention:0, sd:StorageDescriptor(cols:[FieldSchema(name:id, type:int, comment:null), FieldSchema(name:name, type:string, comment:null), FieldSchema(name:age, type:int, comment:null), FieldSchema(name:score, type:float, comment:null)], location:hdfs://xxx-nn1.blue.grid.xxx.com:8020/tmp/xxx/hivedb/student, inputFormat:org.apache.hadoop.mapred.TextInputFormat, outputFormat:org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat, compressed:false, numBuckets:-1, serdeInfo:SerDeInfo(name:null, serializationLib:org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, parameters:{line.delim=

, field.delim= , serialization.format= }), bucketCols:[], sortCols:[], parameters:{}, skewedInfo:SkewedInfo(skewedColNames:[], skewedColValues:[], skewedColValueLocationMaps:{}), storedAsSubDirectories:false), partitionKeys:[], parameters:{hcat.msgbus.topic.name=hcat.jet-blue.students_db.student, totalSize=0, numRows=-1, rawDataSize=-1, COLUMN_STATS_ACCURATE=false, numFiles=0, transient_lastDdlTime=1466792036, comment=Description of the table}, viewOriginalText:null, viewExpandedText:null, tableType:MANAGED_TABLE)

Time taken: 0.048 seconds, Fetched: 7 row(s)

[Read More...]

How to load data from a text file to Hive table

June 24, 2016 Author: david
3 Comments

In this post, I describe how to insert data from a text file to a hive table.

Suppose you have tab delimited file::

% cat /tmp/input.txt k1 v1 k2 v2

1
2
3
4

% cat /tmp/input.txt

k1 v1
k2 v2

Create a Hive table stored as a text file.

create table test_table (k string, v string) </code><code>row format delimited </code><code>fields terminated by '\t' stored as textfile;

1
2
3
4

create table test_table (k string, v string)
</code><code>row format delimited
</code><code>fields terminated by '\t'
stored as textfile;

Load the text file (stored locally) into the Hive table:

hive(test_db)> load data local inpath 'file:/homes/user/hive/input.txt' into table test_table; Loading data to table test_db.test_table OK Time taken: 0.43 seconds

1
2
3
4
5
6
7

hive(test_db)> load data local inpath 'file:/homes/user/hive/input.txt' into table test_table;

Loading data to table test_db.test_table

OK

Time taken: 0.43 seconds

Create a Hive table stored as sequence file.

create table test_sq(k string, v string) stored as sequencefile;

1

create table test_sq(k string, v string) stored as sequencefile;

Now you can load into the sequence table from the text table:

insert into table test_sq select * from test_table;

1

insert into table test_sq select * from test_table;

[Read More...]