Cloudera Data Platformを利⽤して、NiFi / Hive / Impala / Spark を体験できるハンズオンを実施いたします。 データの収集からデータ蓄積、データ分析を通してデータの可視化まで、データ基盤としての基礎となるデータフローを体験することができます。
-
1) NiFi でテストデータを集めて、HDFSにデータを保存
-
2) データETL変換
-
3) HDFSに保存したデータをHUE+Hiveで分析
-
4) HDFSに保存したデータをCML+Sparkで分析
自分のアカウント・サーバーを取得。
サーバーアドレス一覧:Google Docs参照。
皆さんが自分利用しているサーバーに、名前を入力してください。
それぞれAWSで実行している違うVM機です。
CREATE EXTERNAL TABLE customer_csv (
customerid INT,
name STRING,
namekana STRING,
zipcode STRING,
address STRING,
phonenumber STRING,
emailaddress STRING,
sex STRING,
age INT,
frequency INT,
visitedday STRING
) ROW FORMAT
DELIMITED FIELDS TERMINATED BY ','
LINES TERMINATED BY '\n'
LOCATION '/tmp/ftp_input/customer'
;
CREATE TABLE customer STORED AS PARQUET
AS SELECT * FROM customer_csv;
CREATE EXTERNAL TABLE webaccess_log_ext (
host STRING,
ident STRING,
usr STRING,
access_time STRING,
request STRING,
status STRING,
size STRING,
referer STRING,
agent STRING,
cid INT
) ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.RegexSerDe'
WITH SERDEPROPERTIES (
'input.regex' = '([^ ]*) ([^ ]*) ([^ ]*) (-|\\[[^\\]]*\\]) ([^ \"]*|\"[^\"]*\") (-|[0-9]*) (-|[0-9]*)(?: ([^ \"]*|\"[^\"]*\") ([^ \"]*|\"[^\"]*\"))? \"([0-9]*)\"'
)
LOCATION '/tmp/s3_input/webaccess'
;
CREATE TABLE webaccess_log STORED AS PARQUET
AS SELECT * FROM webaccess_log_ext;
-
customerテーブルから住所が千葉県で訪問日時が’2018/06’のデータを検索するSQLです。
-
LIMIT句で検索件数を制御することができます。
SELECT
c.customerid,
c.name,
c.age,
c.address,
c.frequency,
c.visitedday
FROM
customer c
WHERE
(c.age >= 30 and c.age < 40)
AND c.visitedday LIKE '2018-06%'
ORDER BY
c.frequency DESC
LIMIT
10
;
webaccess_logテーブルから2018-07にアクセスがあったデータを取得しています。 毎日のアクセス数を統計します。
SELECT
day,
count(day) as access_count
FROM(
SELECT
substr(access_time,2,10) as day
FROM
webaccess_log
) AS r
WHERE
day LIKE '2018-06%'
GROUP BY day
ORDER BY day
;
SELECT
customerid,
frequency,
(unix_timestamp("2018-07-17", 'yyyy-MM-dd') - unix_timestamp(visitedday, 'yyyy-MM-dd')) / (24 * 60 * 60) as recency,
webrecency
FROM
customer ct
JOIN
(
SELECT
cid,
MIN(unix_timestamp("2018-07-17 +0900", 'yyyy-MM-dd Z') - unix_timestamp(access_time, '[yyyy-MM-dd HH:mm:ss Z]')) / (24 * 60 * 60) AS webrecency
FROM
webaccess_log
GROUP BY
cid
ORDER BY
cid
) AS wl
ON
wl.cid = ct.customerid
// load data
val customer_raw = spark.read.csv("/tmp/ftp_input/customer/*.csv");
customer_raw.show();
val customer = customer_raw.select($"_c0".alias("customerid"), $"_c1".alias("name"), $"_c2".alias("namekana"), $"_c3".alias("zipcode"), $"_c4".alias("address")) ;
customer.show();
// filter : 301-0で始まる行
val filtered = customer.filter ($"zipcode".contains("301-0"));
filtered.show();
// output
filtered.write.parquet("/tmp/filtered_customer");