20220402

pyspark讀寫iceberg# code:utf-8
import findspark
findspark.init(r"D:\Python37\Lib\site-packages\pyspark")
這里要指定pyspark的路徑,如果是服務器的話最好用spark所在的pyspark路徑
import os
java8_location = r'D:\Java\jdk1.8.0_301/'  # 設置你自己的路徑
os.environ['JAVA_HOME'] = java8_location
from pyspark.sql import SparkSessiondef get_spark():# pyspark 讀iceberg表spark = SparkSession.builder.getOrCreate()spark.conf.set("spark.sql.catalog.iceberg", "org.apache.iceberg.spark.SparkCatalog")spark.conf.set("spark.sql.catalog.iceberg.type", "hive")spark.conf.set("spark.sql.catalog.iceberg.uri", "thrift://192.168.1.54:9083")不同的目標地址,不同的服務器集群，要拷貝對應的兩個hive文件到當地客戶端的pyspar conf文件夾下return sparkif __name__ == '__main__':spark = get_spark()pdf = spark.sql("select shangpgg from iceberg.test.end_spec limit 10")spark.sql("insert into iceberg.test.end_spec  values ('aa','bb')")pdf.show()print()
1. 在pyspark下新建conf文件夾，把iceberg下的兩個hive配置文件
放在下面
hdfs-site.xml
hive-site.xm
2. iceberg-spark3-runtime-0.13.1.jar 把這個文件放在pyspark的jars文件夾

20220315

  self.config_iceberg = {"host": "192.168.1.55","port": 8881,"user": "root","catalog": "iceberg","schema": "ice_ods",}import trinoif connected_type == "iceberg":self.conn = trino.dbapi.connect(**self.config_iceberg)iceberg和trino的關系連接

sink寫
source讀

分區數最好和kafkatopic的分區數一樣，否則用默認的200個分區很慢

按天分區相當于一天只有一個目錄

https://blog.csdn.net/xuronghao/article/details/106184831
spark寫入iceberg

partition具體分區


hadoop_prod是具體的catalog,tb是數據庫

通過catalog兩種方式hive或者hadoop來創建數倉

def save_to_db(data,database_type):"""保存至數據庫:param data: 要保存的數據:return: 無返回值"""if database_type == '生產':trino_engine = create_engine("trino://root@192.168.1.55:8881/iceberg/ice_dwt") # 生產庫else:trino_engine = create_engine('trino://root@192.168.40.11:8882/iceberg/ice_dwt')   # 測試庫times = int(np.ceil(data.shape[0] / 1000))for i in tqdm(range(times)):data.iloc[i * 1000 : (i + 1) * 1000, :].to_sql(name="dwt_dm_bi_b2b_customer_churn_wide",con=trino_engine,index=False,if_exists="append",schema="ice_dwt",method="multi",)logger.debug("存入數據庫成功")
可以用此方式插入iceberg雖然速度慢一點

20220314

spark要寫入iceberg需要一個alluxio-2.7.3-client.jar這個jar包
在alluxio下載下來的zip包里面

spark讀寫iceberg沒有測試成功

總結

以上是生活随笔為你收集整理的iceberg问题小结的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。