玩转 Oryx2 (一)

准备

环境

  • CDH 5.5.2, Parcel
    • HDFS
    • YARN
    • Zookeeper
    • Kafka
    • Spark (on YARN)

下载最新版本

  • 进入下载页面,分别下载

    • compute-classpath.sh
    • oryx-batch-2.1.2.jar
    • oryx-run.sh
    • oryx-serving-2.1.2.jar
    • oryx-speed-2.1.2.jar
  • 下载conf文件

    • 以ALS为例,在源码目录oryx/app/conf下als-example.conf

    • 并将其重命名为oryx.conf(注:文件需要和每个层的JAR文件放在同一个目录下)

    • 修改conf文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
  # 现阶段只需要修改这么几个字段就OK了
  kafka-brokers = "data-mining-46.slave:9092,data-mining-47.slave:9092,data-mining-48.slave:9092,data-mining-49.master:9092"
  zk-servers = "data-mining-46.slave:2181,data-mining-47.slave:2181,data-mining-48.slave:2181/kafka"
  hdfs-base = "hdfs:///Oryx"

  oryx {
  id = "ALSExample"
  als {
    rescorer-provider-class = null
  }
  input-topic {
    broker = ${kafka-brokers}
    lock = {
      master = ${zk-servers}
    }
  ...

开始

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
[root@data-mining-49 oryx]# ./oryx-run.sh kafka-setup
Input   ZK      data-mining-46.slave:2181,data-mining-47.slave:2181,data-mining-48.slave:2181/kafka
        Kafka   data-mining-46.slave:9092,data-mining-47.slave:9092,data-mining-48.slave:9092,data-mining-49.master:9092
        topic   OryxInput
Update  ZK      data-mining-46.slave:2181,data-mining-47.slave:2181,data-mining-48.slave:2181/kafka
        Kafka   data-mining-46.slave:9092,data-mining-47.slave:9092,data-mining-48.slave:9092,data-mining-49.master:9092
        topic   OryxUpdate

All available topics:


Input topic OryxInput does not exist. Create it? y
Creating topic OryxInput
Error while executing topic command : org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /brokers/ids
[2016-04-08 16:53:09,383] ERROR org.I0Itec.zkclient.exception.ZkNoNodeException: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode =
NoNode for /brokers/ids
        at org.I0Itec.zkclient.exception.ZkException.create(ZkException.java:47)
        at org.I0Itec.zkclient.ZkClient.retryUntilConnected(ZkClient.java:995)
        at org.I0Itec.zkclient.ZkClient.getChildren(ZkClient.java:675)
        at org.I0Itec.zkclient.ZkClient.getChildren(ZkClient.java:671)
        at kafka.utils.ZkUtils.getChildren(ZkUtils.scala:537)
        at kafka.utils.ZkUtils.getSortedBrokerList(ZkUtils.scala:172)
        at kafka.admin.AdminUtils$.createTopic(AdminUtils.scala:243)
        at kafka.admin.TopicCommand$.createTopic(TopicCommand.scala:107)
        at kafka.admin.TopicCommand$.main(TopicCommand.scala:60)
        at kafka.admin.TopicCommand.main(TopicCommand.scala)
Caused by: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /brokers/ids
        at org.apache.zookeeper.KeeperException.create(KeeperException.java:111)
        at org.apache.zookeeper.KeeperException.create(KeeperException.java:51)
        at org.apache.zookeeper.ZooKeeper.getChildren(ZooKeeper.java:1468)
        at org.apache.zookeeper.ZooKeeper.getChildren(ZooKeeper.java:1496)
        at org.I0Itec.zkclient.ZkConnection.getChildren(ZkConnection.java:114)
        at org.I0Itec.zkclient.ZkClient$4.call(ZkClient.java:678)
        at org.I0Itec.zkclient.ZkClient$4.call(ZkClient.java:675)
        at org.I0Itec.zkclient.ZkClient.retryUntilConnected(ZkClient.java:985)
        ... 8 more
 (kafka.admin.TopicCommand$)
Status of topic OryxInput:
y
Update topic OryxUpdate does not exist. Create it?
Creating topic OryxUpdate
Error while executing topic command : org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /brokers/ids
[2016-04-08 16:53:11,094] ERROR org.I0Itec.zkclient.exception.ZkNoNodeException: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /brokers/ids
        at org.I0Itec.zkclient.exception.ZkException.create(ZkException.java:47)
        at org.I0Itec.zkclient.ZkClient.retryUntilConnected(ZkClient.java:995)
        at org.I0Itec.zkclient.ZkClient.getChildren(ZkClient.java:675)
        at org.I0Itec.zkclient.ZkClient.getChildren(ZkClient.java:671)
        at kafka.utils.ZkUtils.getChildren(ZkUtils.scala:537)
        at kafka.utils.ZkUtils.getSortedBrokerList(ZkUtils.scala:172)
        at kafka.admin.AdminUtils$.createTopic(AdminUtils.scala:243)
        at kafka.admin.TopicCommand$.createTopic(TopicCommand.scala:107)
        at kafka.admin.TopicCommand$.main(TopicCommand.scala:60)
        at kafka.admin.TopicCommand.main(TopicCommand.scala)
Caused by: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /brokers/ids
        at org.apache.zookeeper.KeeperException.create(KeeperException.java:111)
        at org.apache.zookeeper.KeeperException.create(KeeperException.java:51)
        at org.apache.zookeeper.ZooKeeper.getChildren(ZooKeeper.java:1468)
        at org.apache.zookeeper.ZooKeeper.getChildren(ZooKeeper.java:1496)
        at org.I0Itec.zkclient.ZkConnection.getChildren(ZkConnection.java:114)
        at org.I0Itec.zkclient.ZkClient$4.call(ZkClient.java:678)
        at org.I0Itec.zkclient.ZkClient$4.call(ZkClient.java:675)
        at org.I0Itec.zkclient.ZkClient.retryUntilConnected(ZkClient.java:985)
        ... 8 more
 (kafka.admin.TopicCommand$)
Error while executing topic command : Topic OryxUpdate does not exist on ZK path data-mining-46.slave:2181,data-mining-47.slave:2181,data-mining-48.slave:2181/kafka
[2016-04-08 16:53:11,833] ERROR java.lang.IllegalArgumentException: Topic OryxUpdate does not exist on ZK path data-mining-46.slave:2181,data-mining-47.slave:2181,data-mining-48.slave:2181/kafka
        at kafka.admin.TopicCommand$.alterTopic(TopicCommand.scala:119)
        at kafka.admin.TopicCommand$.main(TopicCommand.scala:62)
        at kafka.admin.TopicCommand.main(TopicCommand.scala)
 (kafka.admin.TopicCommand$)
Status of topic OryxUpdate:

[root@data-mining-49 oryx]#

此时出现了一个问题:

1
Error while executing topic command : org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /brokers/ids

这个问题应该是:CDH没有启动Kafka服务没有启动。通过CDH后台启动Kafka服务(如果没有添加Kafka服务,则需要先添加)。

Kafka Service

此时重新执行setup命令:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
[root@data-mining-49 oryx]# ./oryx-run.sh kafka-setup
Input   ZK      data-mining-46.slave:2181,data-mining-47.slave:2181,data-mining-48.slave:2181/kafka
        Kafka   data-mining-47.slave:9092,data-mining-48.slave:9092
        topic   OryxInput
Update  ZK      data-mining-46.slave:2181,data-mining-47.slave:2181,data-mining-48.slave:2181/kafka
        Kafka   data-mining-47.slave:9092,data-mining-48.slave:9092
        topic   OryxUpdate

All available topics:


Input topic OryxInput does not exist. Create it? y
Creating topic OryxInput
Created topic "OryxInput".
Status of topic OryxInput:
Topic:OryxInput PartitionCount:4        ReplicationFactor:1     Configs:
        Topic: OryxInput        Partition: 0    Leader: 140     Replicas: 140   Isr: 140
        Topic: OryxInput        Partition: 1    Leader: 141     Replicas: 141   Isr: 141
        Topic: OryxInput        Partition: 2    Leader: 140     Replicas: 140   Isr: 140
        Topic: OryxInput        Partition: 3    Leader: 141     Replicas: 141   Isr: 141

Update topic OryxUpdate does not exist. Create it? y
Creating topic OryxUpdate
Created topic "OryxUpdate".
Updated config for topic "OryxUpdate".
Status of topic OryxUpdate:
Topic:OryxUpdate        PartitionCount:1        ReplicationFactor:1     Configs:retention.ms=86400000,max.message.bytes=16777216
        Topic: OryxUpdate       Partition: 0    Leader: 140     Replicas: 140   Isr: 140

[root@data-mining-49 oryx]#

OK!

需要将如上的二进制脚本都同步到其他集群的节点上了

1
2
3
4
5
6
7
[root@data-mining-49 data]# scp -r oryx/ root@192.168.1.48:/data/
compute-classpath.sh    100% 1893     1.9KB/s   00:00
oryx-run.sh             100%   13KB  13.2KB/s   00:00
oryx-batch-2.1.2.jar    100%   26MB  26.0MB/s   00:00
oryx-serving-2.1.2.jar  100%   33MB  33.0MB/s   00:01
oryx-speed-2.1.2.jar    100%   26MB  26.0MB/s   00:00
oryx.conf               100%   1884  1.8KB/s    00:00

接下来我们需要找一个可用的数据集:

MovieLens 100K Dataset

将这个数据集的格式修改下:

1
2
3
4
5
6
7
8
9
10
11
12
13
[root@data-mining-49 ml-100k]# tr '\t' ',' < u.data > data.csv
[root@data-mining-49 ml-100k]# tail data.csv
806,421,4,882388897
676,538,4,892685437
721,262,3,877137285
913,209,2,881367150
378,78,3,880056976
880,476,3,880175444
716,204,5,879795543
276,1090,1,874795795
13,225,2,882399156
12,203,3,879959583
[root@data-mining-49 ml-100k]#

数据准备完毕,我们可以通过

1
[root@data-mining-49 oryx]# ./oryx-run.sh kafka-input --input-file data.csv

命令导入系统, 同时打开另一个终端窗口,通过命令

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
[root@data-mining-49 oryx]# ./oryx-run.sh kafka-tail
Input   ZK      data-mining-46.slave:2181,data-mining-47.slave:2181,data-mining-48.slave:2181/kafka
        Kafka   data-mining-47.slave:9092,data-mining-48.slave:9092
        topic   OryxInput
Update  ZK      data-mining-46.slave:2181,data-mining-47.slave:2181,data-mining-48.slave:2181/kafka
        Kafka   data-mining-47.slave:9092,data-mining-48.slave:9092
        topic   OryxUpdate

279,64,1,875308510
646,750,3,888528902
654,370,2,887863914
617,582,4,883789294
913,690,3,880824288
660,229,2,891406212
421,498,4,892241344
495,1091,4,888637503
806,421,4,882388897
676,538,4,892685437
721,262,3,877137285
913,209,2,881367150
378,78,3,880056976
880,476,3,880175444
716,204,5,879795543
276,1090,1,874795795
13,225,2,882399156
12,203,3,879959583

实时跟踪输入导入情况,当数据全部导入完毕后,用户可以手动的通过Ctrl-C关闭这个命令。

如果以上全部成功了,可以关闭这些进程。集群至此已经准备好运行Oryx了。

启动服务

1
2
3
4
./oryx-run.sh batch
./oryx-run.sh speed
./oryx-run.sh serving

blablabla…..好多输出,先不要管他。

确认服务层启动成功以后,可以导入数据(难道刚才导数据是测试用的?)

1
wget --quiet --post-file data.csv --output-document -  --header "Content-Type: text/csv" http://192.168.1.49:8080/ingest

批处理层已经开始触发一个新的计算了。

此时看看HDFS中的数据是这样的:

1
2
3
4
5
6
7
8
9
10
11
12
13
[root@data-mining-49 oryx]# hdfs dfs -ls /Oryx
Found 2 items
drwxr-xr-x   - hdfs supergroup          0 2016-04-08 18:40 /Oryx/data
drwxr-xr-x   - hdfs supergroup          0 2016-04-08 18:40 /Oryx/model
^[[A[root@data-mining-49 oryx]# hdfs dfs -ls /Oryx/data
Found 1 items
drwxr-xr-x   - hdfs supergroup          0 2016-04-08 18:40 /Oryx/data/oryx-1460112000000.data
[root@data-mining-49 oryx]# hdfs dfs -ls /Oryx/model
Found 3 items
drwxr-xr-x   - hdfs supergroup          0 2016-04-08 18:41 /Oryx/model/.checkpoint
drwxr-xr-x   - hdfs supergroup          0 2016-04-08 18:40 /Oryx/model/.temporary
drwxr-xr-x   - hdfs supergroup          0 2016-04-08 18:40 /Oryx/model/1460112018440
[root@data-mining-49 oryx]#

表明数据已经ok了,通过API确认计算是否完成

1
2
3
4
5
[root@data-mining-49 ~]# wget --quiet --output-document - --server-response http://192.168.1.49:8080/ready
  HTTP/1.1 200 OK
  Content-Length: 0
  Date: Fri, 08 Apr 2016 10:43:27 GMT
  Server: Oryx

OK!

我们来查看下推荐结果

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
[root@data-mining-49 ~]# wget --quiet --output-document -  http://192.168.1.49:8080/recommend/17
50,0.8234792673029006
127,0.7889597890898585
181,0.7442612769082189
275,0.7263787714764476
258,0.7164891492575407
121,0.7006391943432391
15,0.697495789732784
288,0.6787936894688755
285,0.6696474105119705
25,0.6603603088587988
[root@data-mining-49 ~]# wget --quiet --output-document -  http://192.168.1.49:8080/recommend/806
173,1.084704713546671
64,1.0142679414711893
7,0.9939510896801949
183,0.9918235178338364
22,0.9644620651379228
69,0.945814429782331
202,0.9420167971402407
11,0.9270770070143044
135,0.9227914617804345
191,0.9215727103874087
[root@data-mining-49 ~]#

完美结束。

这貌似是走的批处理逻辑,中途实时计算层挂鸟

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
16/04/08 18:42:38 INFO SparkContext: Successfully stopped SparkContext
16/04/08 18:42:38 INFO OutputCommitCoordinator$OutputCommitCoordinatorEndpoint: OutputCommitCoordinator stopped!
Exception in thread "main" org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 0.0 failed 4 times, most recent failure: Lost task 0.3 in stage 0.0 (TID 3, data-mining-47.slave): java.io.IOException: org.apache.spark.SparkException: Failed to get broadcast_0_piece0 of broadcast_0
        at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1177)
        at org.apache.spark.broadcast.TorrentBroadcast.readBroadcastBlock(TorrentBroadcast.scala:165)
        at org.apache.spark.broadcast.TorrentBroadcast._value$lzycompute(TorrentBroadcast.scala:64)
        at org.apache.spark.broadcast.TorrentBroadcast._value(TorrentBroadcast.scala:64)
        at org.apache.spark.broadcast.TorrentBroadcast.getValue(TorrentBroadcast.scala:88)
        at org.apache.spark.broadcast.Broadcast.value(Broadcast.scala:70)
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:62)
        at org.apache.spark.scheduler.Task.run(Task.scala:88)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
        at java.lang.Thread.run(Thread.java:745)
Caused by: org.apache.spark.SparkException: Failed to get broadcast_0_piece0 of broadcast_0
        at org.apache.spark.broadcast.TorrentBroadcast$$anonfun$org$apache$spark$broadcast$TorrentBroadcast$$readBlocks$1$$anonfun$2.apply(TorrentBroadcast.scala:138)
        at org.apache.spark.broadcast.TorrentBroadcast$$anonfun$org$apache$spark$broadcast$TorrentBroadcast$$readBlocks$1$$anonfun$2.apply(TorrentBroadcast.scala:138)
        at scala.Option.getOrElse(Option.scala:120)

咱们下一篇再找找原因吧。

参考文献

  • http://oryx.io/docs/admin.html
  • http://oryx.io/docs/endusers.html
  • http://oryx.io/docs/developer.html
  • http://oryx.io/docs/performance.html
  • http://oryx.io/apidocs/index.html