大数据生态集成

import org.apache.spark.sql.SparkSession
 
val spark = SparkSession.builder()
  .appName("ProtonBase Integration")
  .getOrCreate()
 
// 通过 JDBC 连接 ProtonBase
val df = spark.read
  .format("jdbc")
  .option("url", "jdbc:postgresql://protonbase-host:5432/database")
  .option("dbtable", "analytics_table")
  .option("user", "username")
  .option("password", "password")
  .load()
 
// 执行数据分析
df.groupBy("category").count().show()

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 
JDBCConnectionOptions jdbcConnectionOptions = new JDBCConnectionOptions.JDBCConnectionOptionsBuilder()
    .withUrl("jdbc:postgresql://protonbase-host:5432/database")
    .withDriverName("org.postgresql.Driver")
    .withUsername("username")
    .withPassword("password")
    .build();
 
JDBCInputFormat jdbcInputFormat = JDBCInputFormat.buildJDBCInputFormat()
    .setDrivername("org.postgresql.Driver")
    .setDBUrl("jdbc:postgresql://protonbase-host:5432/database")
    .setUsername("username")
    .setPassword("password")
    .setQuery("SELECT * FROM streaming_data")
    .setRowTypeInfo(new RowTypeInfo(Types.INT, Types.STRING))
    .finish();
 
DataSet<Row> source = env.createInput(jdbcInputFormat);