利用ogg实现oracle到kafka的增量数据实时同步

2018-05-23

前言

ogg即Oracle GoldenGate是Oracle的同步工具，本文讲如何配置ogg以实现Oracle数据库增量数据实时同步到kafka中，其中同步消息格式为json。
下面是我的源端和目标端的一些配置信息：

	版本	OGG版本	ip	别名
源端	OracleRelease 11.2.0.1.0	Oracle GoldenGate 11.2.1.0.3 for Oracle on Linux x86-64	192.168.44.128	master
目标端	kafka_2.11-1.1.0	Oracle GoldenGate for Big Data 12.3.1.1.1 on Linux x86-64	192.168.44.129	slave1

1、下载

可在这里或旧版本查询下载
注意：源端和目标端的文件不一样，目标端需要下载Oracle GoldenGate for Big Data,源端需要下载Oracle GoldenGate for Oracle具体下载方法见最后的附录截图。

2、源端（Oracle）配置

注意：源端是安装了oracle的机器，oracle环境变量之前都配置好了

2.1 解压

先建立ogg目录

1 2	mkdir -p /opt/ogg unzip V34339-01.zip

解压后得到一个tar包，再解压这个tar

1 2	tar xf fbo_ggs_Linux_x64_ora11g_64bit.tar -C /opt/ogg chown -R oracle:oinstall /opt/ogg （使oracle用户有ogg的权限，后面有些需要在oracle用户下执行才能成功）

more >>

展开全文 >>

Kafka安装启动入门教程

2018-05-21

前言

本文讲如何安装启动kafka,并进行测试，其中zookeepr是kafka自带的，本文基本按照官网文档进行安装启动的，并提出可能会出现的问题。官方文档：http://kafka.apache.org/quickstart
本文虚拟机系统：centos7，不过其他版本的Linux系统是一样的~

1、下载

可直接在官网下载对应的版本http://kafka.apache.org/downloads，我下载的是二进制版的，由于我的scala版本是2.11，所以下载kafka_2.11-1.1.0.tgz，大家可以根据自己的实际情况选择对应的版本。执行以下命令即可下载到本地了。

1	wget http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/1.1.0/kafka_2.11-1.1.0.tgz

2、解压到指定目录

1	tar -xzf kafka_2.11-1.1.0.tgz -C /opt/

3、启动服务

3.1 启动zookeeper

kafka用到zookeeper，因此如果您的机器上没有zookeeper服务，则需要先启动zookpeer服务，本文使用kafka自带的zookeeper。

1 2	cd /opt/kafka_2.11-1.1.0/ bin/zookeeper-server-start.sh config/zookeeper.properties

more >>

展开全文 >>

hive查询报错:java.io.IOException:org.apache.parquet.io.ParquetDecodingException

2018-05-20

前言

本文解决如标题所述的一个hive查询异常，详细异常信息为：

Failed with exception java.io.IOException:org.apache.parquet.io.ParquetDecodingException: Can not read value at 1 in block 0 in file hdfs://192.168.44.128:8888/user/hive/warehouse/test.db/test/part-00000-9596e4bd-f511-4f76-9030-33e426d0369c-c000.snappy.parquet

这个异常是用spark sql将oracle（不知道mysql中有没有该问题，大家可以自己测试一下）中表数据查询出来然后写入hive表中，之后在hive命令行执行查询语句时产生的，下面先具体看一下如何产生这个异常的。

1、建立相关的库和表

1.1 建立hive测试库

在hive里执行如下语句

1	create database test;

1.2 建立oracle测试表

CREATE TABLE TEST
(	"ID" VARCHAR2(100), 
	"NUM" NUMBER(10,2)
)

1.3 在oracle表里插入一条记录

1	INSERT INTO TEST (ID, NUM) VALUES('1', 1);

more >>

展开全文 >>

Spark Streaming连接Kafka入门教程

2018-05-17

前言

首先要安装好kafka,这里不做kafka安装的介绍(这里用的是ambari安装的kafka),若想了解如何安装可参考Kafka安装启动入门教程和centos7 ambari2.6.1.5+hdp2.6.4.0 大数据集群安装部署,本文是Spark Streaming入门教程，只是简单的介绍如何利用spark 连接kafka，并消费数据，由于博主也是才学，所以其中代码以实现为主，可能并不是最好的实现方式。

1、对应依赖

根据kafka版本选择对应的依赖，我的kafka版本为0.10.1，spark版本2.2.1,然后在maven仓库找到对应的依赖。
（Kafka项目在版本0.8和0.10之间引入了新的消费者API，因此有两个独立的相应Spark Streaming软件包可用）

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
    <version>2.2.1</version>
</dependency>

我用的是sbt，对应的依赖：

1	"org.apache.spark" % "spark-streaming-kafka-0-10_2.11" % "2.2.1"

more >>

展开全文 >>

spark ML之特征处理（1）

2018-05-17

前言

最近在学习总结机器学习常用算法，在看spark机器学习决策树的官方示例时，发现用到了几个特征处理的类，之前没学习过，所以查了一下，感觉spark在特征处理方面的类还是挺多的，所以准备总结记录一下相关的用法，首先总结一下决策树中用到的几种。

1、VectorIndexer

根据源码注释，VectorIndexer是用于在“向量”的数据集中索引分类特征列的类（Class for indexing categorical feature columns in a dataset of Vector），这看起来不太好理解，直接看用法，举例说明就好了。

1.1 数据

我们用普通的数据格式即可：
data1.txt

1,-1.0 1.0
0,0.0 3.0
1,-1.0 5.0
0,0.0 1.0

其中第一列为label,后面的为features
spark读取数据程序（供参考）：

import spark.implicits._
val data_path = "files/ml/featureprocessing/data1.txt"
val data = spark.read.text(data_path).map {
  case Row(line: String) =>
    var arr = line.split(',')
    (arr(0), Vectors.dense(arr(1).split(' ').map(_.toDouble)))
}.toDF("label", "features")
data.show(false)

more >>

展开全文 >>