Uber永久定位系统实时数据分析过程实践！

接下来，为了进行调试，我们可以开始接收数据并将数据作为内存表存储在内存中，然后进行查询。

以下是来自 %sqlselect * from uber limit 10 的示例输出：

现在我们可以查询流数据，询问哪段时间和集群内的搭乘次数最多?(输出显示在Zeppelin notebook中)

%sql 

SELECT hour(uber.dt) as hr,cid, count(cid) as ct FROM uber group By hour(uber.dt), cid

Spark Streaming写入MapR-DB

用于Apache Spark的MapR-DB连接器使用户可以将MapR-DB用作Spark结构化流或Spark Streaming的接收器。

当你处理大量流数据时，其中一个挑战是存储位置。对于此应用程序，可以选择MapR-DB JSON(一种高性能NoSQL数据库)，因为它具有JSON的可伸缩性和灵活易用性。

JSON模式的灵活性

MapR-DB支持JSON文档作为本机数据存储。MapR-DB使用JSON文档轻松存储，查询和构建应用程序。Spark连接器可以轻松地在JSON数据和MapR-DB之间构建实时或批处理管道，并在管道中利用Spark。

使用MapR-DB，表按集群的键范围自动分区，提供可扩展行和快速读写能力。在此用例中，行键_id由集群ID和反向时间戳组成，因此表将自动分区，并按最新的集群ID进行排序。

Spark MapR-DB Connector利用Spark DataSource API。连接器体系结构在每个Spark Executor中都有一个连接对象，允许使用MapR-DB(分区)进行分布式并行写入，读取或扫描。

写入MapR-DB接收器

要将Spark Stream写入MapR-DB，请使用tablePath，idFieldPath，createTable，bulkMode和sampleSize参数指定格式。以下示例将cdf DataFrame写到MapR-DB并启动流。

使用Spark SQL查询MapR-DB JSON

Spark MapR-DB Connector允许用户使用Spark数据集在MapR-DB之上执行复杂的SQL查询和更新，同时应用投影和过滤器下推，自定义分区和数据位置等关键技术。

将数据从MapR-DB加载到Spark数据集中

要将MapR-DB JSON表中的数据加载到Apache Spark数据集，我们可在SparkSession对象上调用loadFromMapRDB方法，提供tableName，schema和case类。这将返回UberwId对象的数据集：

使用Spark SQL探索和查询Uber数据

现在，我们可以查询连续流入MapR-DB的数据，使用Spark DataFrames特定于域的语言或使用Spark SQL来询问。

显示第一行(注意行如何按_id分区和排序，_id由集群ID和反向时间戳组成，反向时间戳首先排序最近的行)。

df.show 

每个集群发生多少次搭乘?

df.groupBy("cid").count().orderBy(desc( "count")).show 

（编辑：萍乡站长网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!