import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
//該類展示如何使用Java將本地數據存儲到HDFS中
public class HDFSDataStorage {
public static void main(String[] args) throws Exception {
//創建Hadoop配置對象
Configuration conf = new Configuration();
//設置HDFS的默認地址
conf.set("fs.defaultFS", "hdfs://localhost:9000");
//獲取HDFS文件系統實例
FileSystem fs = FileSystem.get(conf);
//將本地文件複製到HDFS中
fs.copyFromLocalFile(new Path("local_data.txt"), new Path("/hdfs_data.txt"));
//關閉文件系統資源
fs.close();
}
}

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
//該類展示如何使用Java向HBase中寫入數據
public class HBaseDataWriter {
public static void main(String[] args) {
try {
//創建HBase配置對象
Configuration conf = HBaseConfiguration.create();
//創建HBase連接
Connection connection = ConnectionFactory.createConnection(conf);
//獲取表對象
Table table = connection.getTable(TableName.valueOf("electricity_market"));
//創建Put對象，指定行鍵
Put put = new Put(Bytes.toBytes("row1"));
//添加列數據
put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("col1"), Bytes.toBytes("value1"));
//將數據寫入表中
table.put(put);
//關閉表和連接
table.close();
connection.close();
} catch (Exception e) {
e.printStackTrace();
}
}
}

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.mllib.fpm.AssociationRules;
import org.apache.spark.mllib.fpm.FPGrowth;
import org.apache.spark.mllib.fpm.FPGrowthModel;
import java.util.ArrayList;
import java.util.List;
//該類展示如何使用Spark進行關聯規則挖掘
public class SparkAssociationRules {
public static void main(String[] args) {
//創建Spark配置對象，設置應用名稱和運行模式
SparkConf conf = new SparkConf().setAppName("SparkAssociationRules").setMaster("local[*]");
//創建Java Spark上下文對象
JavaSparkContext sc = new JavaSparkContext(conf);
List<List<String>> transactions = new ArrayList<>();
  //模擬原始交易數據
  String rawData = "item1,item2;item2,item3";
  for (String transaction : rawData.split(";")) {
  List<String> items = new ArrayList<>();
    for (String item : transaction.split (",")) {
    items.add(item);
    }
    transactions.add(items);
    }
    //將交易數據轉換為Spark RDD
    JavaRDD<List<String>> rdd = sc.parallelize(transactions);
      //創建FPGrowth對象，設置最小支持度和分區數
      FPGrowth fpg = new FPGrowth()
      .setMinSupport(0.2)
      .setNumPartitions(10);
      //運行FPGrowth算法，生成關聯規則模型
      FPGrowthModel<String> model = fpg.run(rdd);
        //遍歷生成的關聯規則，並打印規則前件、後件和置信度
        for (AssociationRules.Rule<String> rule : model.generateAssociationRules(0.8)) {
          System.out.println(rule.javaAntecedent() + " => " + rule.javaConsequent() + ", " + rule.confidence());
          }
          //停止Spark上下文
          sc.stop();
          }
          }

import weka.core.Attribute;
import weka.core.DenseInstance;
import weka.core.Instance;
import weka.core.Instances;
import weka.regression.LinearRegression;
import weka.core.converters.ConverterUtils.DataSource;
//該類展示如何使用Weka庫進行電力市場交易價格預測
public class PricePrediction {
public static void main(String[] args) throws Exception {
// 加載數據集
DataSource source = new DataSource("electricity_price_data.arff");
Instances dataset = source.getDataSet();
dataset.setClassIndex(dataset.numAttributes() - 1);
// 分割數據集為訓練集和測試集
Instances train = dataset.trainCV(8, 0);
Instances test = dataset.testCV(8, 0);
//創建線性迴歸對象
LinearRegression lr = new LinearRegression();
//使用訓練數據構建線性迴歸模型
lr.buildClassifier(train);
// 評估模型
double sumError = 0;
for (int i = 0; i < test.numInstances(); i++) {
double predicted = lr.classifyInstance(test.instance(i));
double actual = test.instance(i).classValue();
sumError += Math.abs(predicted - actual);
}
double averageError = sumError / test.numInstances();
System.out.println("Average Error: " + averageError);
// 創建新的實例，用於預測
Instance newInstance = new DenseInstance(3);
newInstance.setValue((Attribute) dataset.attribute(0), 4.0);
newInstance.setValue((Attribute) dataset.attribute(1), 5.0);
// 預測價格
double predictedPrice = lr.classifyInstance(newInstance);
System.out.println("Predicted Price: " + predictedPrice);
}
}