Tracking:
-
安裝schema register:
該服務用來注冊pinot需要的scheme吮螺，這樣所有需要schema的地方調用該服務即可. http://docs.confluent.io/1.0/installation.html#installation 修改兼容性: curl -X PUT -i -H "Content-Type: application/vnd.schemaregistry.v1+json" --data '{"compatibility": "FORWARD"}' http://localhost:8801{"compatibility": "FORWARD"} 如何啟動: schema-registrybin/schema-registry-start config/schema-registry.properties & 在 schema-registry 服務中創(chuàng)建AvroSchema 1. Register a new version of a schema under the subject "WechatActiveUsersV3" $ curl -X POST -i -H "Content-Type: application/vnd.schemaregistry.v1+json" --data '{"schema": "{\"type\": \"string\"}"}' http://localhost:8081/subjects/Kafka-key/versions 2. List all subjects $ curl -X GET -i http://localhost:8081/subjects 3. List all schema versions registered under the subject "WechatActiveUsersV3" $ curl -X GET -i http://localhost:8081/subjects/WechatActiveUsersV3/versions 4. Fetch the most recently registered schema under subject "Kafka-value" $ curl -X GET -i http://localhost:8081/subjects/Kafka-value/versions/latest 5. Update compatibility requirements globally $ curl -X PUT -i -H "Content-Type: application/vnd.schemaregistry.v1+json" --data '{"compatibility": "NONE"}' http://localhost:8081/config 6. Update compatibility requirements under the subject "Kafka-value"$ curl -X PUT -i -H "Content-Type: application/vnd.schemaregistry.v1+json" --data '{"compatibility": "BACKWARD"}' http://localhost:8081/config/Kafka-value 執(zhí)行Pinot索引創(chuàng)建 $ gobblin-mapreduce.sh --conf job-conf/WechaActiveUserV3.pull --jars target/parllay-pipeline-gobblin-1.0.0.jar $ hadoop jar /opt/pinot-0.016-pkg/lib/pinot-hadoop-0.016.jar SegmentCreation /path/jobs/WechatActiveUsersV3-Job.properties $ hadoop jar /opt/pinot-0.016-pkg/lib/pinot-hadoop-0.016.jar SegmentTarPush /path/*-Job.properties 創(chuàng)建Pinot對應的Avro元數(shù)據(jù) $ curl -X POST -i -H "Content-Type: application/vnd.schemaregistry.v1+json" --data '{"schema":"{\"namespace\":\"com.parllay.wechat\",\"type\":\"record\",\"name\":\"wechatActiveUserMetrics\",\"fields\":[{\"name\":\"platId\",\"type\":\"string\"},{\"name\":\"openId\",\"type\":\"string\"},{\"name\":\"year\",\"type\":\"int\"},{\"name\":\"month\",\"type\":\"int\"},{\"name\":\"week\",\"type\":\"int\"},{\"name\":\"day\",\"type\":\"int\"},{\"name\":\"hour\",\"type\":\"int\"},{\"name\":\"counter\",\"type\":\"int\"},{\"name\":\"eventTime\",\"type\":\"long\"}]}"}' http://schema.registry:8801/subjects/wechatActiveUserMetrics/versions $ curl -X POST -i -H "Content-Type: application/vnd.schemaregistry.v1+json" --data '{"schema":"{\"namespace\":\"com.parllay.wechat\",\"type\":\"record\",\"name\":\"wechatEventMetrics\",\"fields\":[{\"name\":\"platId\",\"type\":\"string\"},{\"name\":\"event\",\"type\":\"string\"},{\"name\":\"eventKey\",\"type\":\"string\"},{\"name\":\"eventValue\",\"type\":\"string\"},{\"name\":\"eventKey1\",\"type\":\"string\"},{\"name\":\"eventValue1\",\"type\":\"string\"},{\"name\":\"eventKey2\",\"type\":\"string\"},{\"name\":\"eventValue2\",\"type\":\"string\"},{\"name\":\"countryId\",\"type\":\"int\"},{\"name\":\"provinceId\",\"type\":\"int\"},{\"name\":\"cityId\",\"type\":\"int\"},{\"name\":\"realTimeCountryId\",\"type\":\"int\"},{\"name\":\"realTimeProvinceId\",\"type\":\"int\"},{\"name\":\"realTimeCityId\",\"type\":\"int\"},{\"name\":\"gender\",\"type\":\"int\"},{\"name\":\"groupId\",\"type\":\"int\"},{\"name\":\"languageId\",\"type\":\"int\"},{\"name\":\"tags\",\"type\":{\"type\":\"array\",\"items\":\"int\"}},{\"name\":\"year\",\"type\":\"int\"},{\"name\":\"month\",\"type\":\"int\"},{\"name\":\"week\",\"type\":\"int\"},{\"name\":\"day\",\"type\":\"int\"},{\"name\":\"hour\",\"type\":\"int\"},{\"name\":\"counter\",\"type\":\"int\"},{\"name\":\"eventTime\",\"type\":\"long\"}]}"}' http://schema.registry:8801/subjects/wechatEventMetrics/versions $ curl -X POST -i -H "Content-Type: application/vnd.schemaregistry.v1+json" --data '{"schema":"{\"namespace\":\"com.parllay.wechat\",\"type\":\"record\",\"name\":\"wechatUserMetrics\",\"fields\":[{\"name\":\"platId\",\"type\":\"string\"},{\"name\":\"openId\",\"type\":\"string\"},{\"name\":\"event\",\"type\":\"string\"},{\"name\":\"eventKey\",\"type\":\"string\"},{\"name\":\"eventValue\",\"type\":\"string\"},{\"name\":\"eventKey1\",\"type\":\"string\"},{\"name\":\"eventValue1\",\"type\":\"string\"},{\"name\":\"eventKey2\",\"type\":\"string\"},{\"name\":\"eventValue2\",\"type\":\"string\"},{\"name\":\"countryId\",\"type\":\"int\"},{\"name\":\"provinceId\",\"type\":\"int\"},{\"name\":\"cityId\",\"type\":\"int\"},{\"name\":\"realTimeCountryId\",\"type\":\"int\"},{\"name\":\"realTimeProvinceId\",\"type\":\"int\"},{\"name\":\"realTimeCityId\",\"type\":\"int\"},{\"name\":\"gender\",\"type\":\"int\"},{\"name\":\"groupId\",\"type\":\"int\"},{\"name\":\"languageId\",\"type\":\"int\"},{\"name\":\"tags\",\"type\":{\"type\":\"array\",\"items\":\"int\"}},{\"name\":\"year\",\"type\":\"int\"},{\"name\":\"month\",\"type\":\"int\"},{\"name\":\"week\",\"type\":\"int\"},{\"name\":\"day\",\"type\":\"int\"},{\"name\":\"hour\",\"type\":\"int\"},{\"name\":\"counter\",\"type\":\"int\"},{\"name\":\"eventTime\",\"type\":\"long\"}]}"}' http://schema.registry:8801/subjects/wechatUserMetrics/versions -
安裝hadoop集群：
啟動該服務hadoop-2.7.1 -
安裝spark集群：
啟動該服務spark2.0.1 安裝azkaban:
啟動該服務,詳細安裝過程參見我的上一篇文章
http://www.reibang.com/p/fd187cb15674-
編譯pinot:
啟動controller, broker, server bin/start-server.sh -zkAddress=master:2181 -clusterName=pinot-fht-cluster-v2 -serverHost=slave1 -serverPort=8098 -dataDir=/opt/pinot-0.016-pkg/data -segmentDir=/opt/pinot-0.016-pkg/segment & bin/start-broker.sh -zkAddress=master:2181 -clusterName=pinot-fht-cluster-v2 -brokerHost=slave1 -brokerPort=8099 & bin/start-controller.sh -zkAddress=master:2181 -clusterName=pinot-fht-cluster-v2 -controllerHost=slave1 -controllerPort=9001 -dataDir=/pinot-0.016-pkg/controller & -
編譯gobblin:
從kafka導出數(shù)據(jù)到hdfs, 每次導出都會在zookeeper記錄其offset及其wartermark. 每次執(zhí)行的數(shù)據(jù)段為上次offset到gobblin啟動任務的時間段. -
安裝kafka集群
啟動集群kafka_2.11-0.9.0.1旷赖，創(chuàng)建topic
創(chuàng)建tracking schema, table等
提交schema到schema register服務.
produce數(shù)據(jù)到kafka集群
測試pinot 數(shù)據(jù)接收情況. 根據(jù)web控制臺
生成離線數(shù)據(jù).
測試整個數(shù)據(jù)流程.
