# 示例：安装Oracle Big Data SQL

# 1. 下载Oracle Big Data SQL
– 从Oracle官网下载Oracle Big Data SQL

# 2. 安装Oracle Big Data SQL
$ ./runInstaller.sh

# 3. 配置Hadoop连接
– 编辑配置文件：/opt/oracle/bigdatasql/bds-config.ini
– 设置Hadoop集群信息

# 4. 创建外部表

# 创建外部表指向Hadoop数据
CREATE TABLE hadoop_fgfgfgsales (
sale_id NUMBER,
product_id NUMBER,
customer_id NUMBER,
sale_date DATE,
amount NUMBER
) ORGANIZATION EXTERNAL (
TYPE ORACLE_HADOOP
DEFAULT DIRECTORY hadoop_dir
ACCESS PARAMETERS (
com.oracle.bigdata.fileformat = ‘ORC’,
com.oracle.bigdata.tablespace = ‘default’,
com.oracle.bigdata.tablename = ‘fgfgfgsales’
)
) PARALLEL 8;

# 5. 查询Hadoop数据

# 查询Hadoop数据
SELECT * FROM hadoop_fgfgfgsales WHERE sale_date > ‘2023-01-01’;

# 联合查询Oracle和Hadoop数据
SELECT o.customer_name, h.amount
FROM oracle_customers o
JOIN hadoop_fgfgfgsales h ON o.customer_id = h.customer_id
WHERE h.sale_date > ‘2023-01-01’;

# 聚合查询
SELECT product_id, SUM(amount) AS total_fgfgfgsales
FROM hadoop_fgfgfgsales
GROUP BY product_id
ORDER BY total_fgfgfgsales DESC;

# 6. 优化查询性能

# 创建外部表索引
CREATE INDEX hadoop_fgfgfgsales_date_idx ON hadoop_fgfgfgsales(sale_date)
INDEXTYPE IS ctxsys.context;

# 使用并行查询
ALTER SESSION ENABLE PARALLEL DML;
ALTER SESSION FORCE PARALLEL QUERY PARALLEL 8;

# 示例：使用Oracle Data Integrator集成Hadoop

# 1. 配置ODI连接到Hadoop
– 创建Hadoop数据源
– 配置连接参数

# 2. 创建映射
– 源：Hadoop Hive表
– 目标：Oracle表
– 转换：数据转换规则

# 3. 执行集成
– 运行ODI包
– 监控执行状态
– 验证数据

# 4. 增量集成
– 配置增量提取
– 执行增量加载
– 验证增量数据

# 示例：安装Oracle Spark Connector

# 1. 下载Oracle Spark Connector
– 从Oracle官网下载Oracle Spark Connector

# 2. 配置Spark环境
– 将Oracle Spark Connector JAR文件复制到Spark lib目录
– 配置Spark配置文件

# 3. 读取Oracle数据

# Spark Scala代码
val spark = SparkSession.builder()
.appName(“Oracle Spark Integration”)
.getOrCreate()

// 读取Oracle数据
val oracleDF = spark.read
.format(“oracle”)
.option(“url”, “jdbc:oracle:thin:@localhost:1521:ORCL”)
.option(“dbtable”, “scott.emp”)
.option(“user”, “scott”)
.option(“password”, “tiger”)
.load()

// 显示数据
oracleDF.show()

# 4. 处理数据

// 数据处理
val processedDF = oracleDF.filter($”sal” > 2000)
.groupBy($”deptno”)
.agg(avg($”sal”).as(“avg_sal”))

// 显示处理结果
processedDF.show()

# 5. 写入数据回Oracle

// 写入Oracle
processedDF.write
.format(“oracle”)
.option(“url”, “jdbc:oracle:thin:@localhost:1521:ORCL”)
.option(“dbtable”, “scott.dept_avg_sal”)
.option(“user”, “scott”)
.option(“password”, “tiger”)
.mode(“overwrite”)
.save()

# 示例：使用Spark SQL查询Oracle数据

// 创建临时表
oracleDF.createOrReplaceTempView(“emp”)

// 执行SQL查询
val resultDF = spark.sql(“””
SELECT deptno, AVG(sal) as avg_sal
FROM emp
WHERE sal > 2000
GROUP BY deptno
ORDER BY avg_sal DESC
“””)

// 显示结果
resultDF.show()

# 示例：使用Spark Streaming处理实时数据

// 创建StreamingContext
val ssc = new StreamingContext(spark.sparkContext, Seconds(5))

// 从Kafka读取数据
val kafkaStream = KafkaUtils.createDirectStream[String, String](
ssc,
PreferConsistent,
Subscribe[String, String]([“fgfgfgsales”], kafkaParams)
)

// 处理数据
val processedStream = kafkaStream.map(record => {
val value = record.value()
// 解析JSON数据
val json = JSON.parseFull(value).get.asInstanceOf[Map[String, Any]]
(json(“product_id”), json(“amount”))
})

// 聚合数据
val fgfgfgsalesByProduct = processedStream.mapValues(_.toDouble)
.reduceByKey(_ + _)

// 写入Oracle
fgfgfgsalesByProduct.foreachRDD(rdd => {
rdd.foreachPartition(partition => {
// 创建数据库连接
val conn = DriverManager.getConnection(“jdbc:oracle:thin:@localhost:1521:ORCL”, “scott”, “tiger”)
partition.foreach { case (productId, amount) => {
// 插入数据
val stmt = conn.prepareStatement(“INSERT INTO fgfgfgsales (product_id, amount, sale_date) VALUES (?, ?, SYSDATE)”)
stmt.setInt(1, productId.toString.toInt)
stmt.setDouble(2, amount)
stmt.executeUpdate()
stmt.close()
}}
conn.close()
})
})

// 启动流处理
ssc.start()
ssc.awaitTermination()

# 示例：安装Oracle NoSQL Database

# 1. 下载Oracle NoSQL Database
– 从Oracle官网下载Oracle NoSQL Database

# 2. 安装Oracle NoSQL Database
$ java -jar nosql-db-ee-23.3.12.jar

# 3. 配置NoSQL数据库
$ java -jar kvstore-23.3.12.jar makebootconfig -root /opt/oracle/nosql -host localhost -port 5000 -store kvstore

# 4. 启动NoSQL数据库
$ java -jar kvstore-23.3.12.jar start -root /opt/oracle/nosql

# 5. 连接NoSQL数据库

# 使用KVStore API连接
import oracle.kv.KVStore;
import oracle.kv.KVStoreConfig;
import oracle.kv.KVStoreFactory;

KVStore store = KVStoreFactory.getStore(
new KVStoreConfig(“kvstore”, “localhost:5000”)
);

# 6. 写入NoSQL数据

// 写入数据
import oracle.kv.Key;
import oracle.kv.Value;

Key key = Key.createKey(Arrays.asList(“customer”, “1001”));
Value value = Value.createValue(“{\”name\”: \”John Doe\”, \”email\”: \”john.doe@fgedu.net.cn\”}”.getBytes());
store.put(key, value);

# 7. 读取NoSQL数据

// 读取数据
Value value = store.get(key);
if (value != null) {
String json = new String(value.getValue());
System.out.println(json);
}

# 8. 与Oracle数据库集成

# 使用Oracle External Tables
CREATE TABLE nosql_customers (
customer_id VARCHAR2(10),
customer_data CLOB
) ORGANIZATION EXTERNAL (
TYPE ORACLE_HDFS
DEFAULT DIRECTORY nosql_dir
ACCESS PARAMETERS (
com.oracle.bigdata.fileformat = ‘JSON’,
com.oracle.bigdata.tablespace = ‘default’,
com.oracle.bigdata.tablename = ‘customers’
)
);

# 查询NoSQL数据
SELECT customer_id, customer_data
FROM nosql_customers
WHERE JSON_VALUE(customer_data, ‘$.name’) = ‘John Doe’;

# 示例：使用Oracle NoSQL Database Cloud Service

# 1. 创建NoSQL Cloud Service实例
– 登录Oracle Cloud Console
– 创建NoSQL Database Cloud Service实例

# 2. 配置连接
– 获取连接字符串
– 配置认证信息

# 3. 连接NoSQL Cloud Service

// 使用Java SDK连接
import com.oracle.nosql.driver.kv.NoSQLHandle;
import com.oracle.nosql.driver.kv.NoSQLHandleConfig;
import com.oracle.nosql.driver.kv.StoreAccessTokenProvider;

NoSQLHandleConfig config = new NoSQLHandleConfig(“https://nosql.us-east-1.oci.oraclecloud.com”);
config.setAccessTokenProvider(new StoreAccessTokenProvider());
NoSQLHandle handle = NoSQLHandleFactory.createNoSQLHandle(config);

# 4. 读写数据

// 写入数据
MapValue value = new MapValue()
.put(“name”, new StringValue(“John Doe”))
.put(“email”, new StringValue(“john.doe@fgedu.net.cn”));

PutRequest putRequest = new PutRequest()
.setTableName(“customers”)
.setValue(value)
.setReturnRow(true);

PutResult putResult = handle.put(putRequest);

// 读取数据
GetRequest getRequest = new GetRequest()
.setTableName(“customers”)
.setKey(Arrays.asList(new IntegerValue(1001)));

GetResult getResult = handle.get(getRequest);
if (getResult.getRow() != null) {
System.out.println(getResult.getRow());
}