import pandas as pd
import sqlite3

# 創(chuàng)建數(shù)據(jù)庫連接
conn = sqlite3.connect('data_warehouse.db')
cursor = conn.cursor()

# 創(chuàng)建事實(shí)表與維度表
cursor.execute('''CREATE TABLE IF NOT EXISTS fact_sales (
                    sale_id INTEGER PRIMARY KEY,
                    product_id INTEGER,
                    customer_id INTEGER,
                    sales_amount REAL,
                    sale_date TEXT)''')

cursor.execute('''CREATE TABLE IF NOT EXISTS dim_product (
                    product_id INTEGER PRIMARY KEY,
                    product_name TEXT,
                    category TEXT)''')

cursor.execute('''CREATE TABLE IF NOT EXISTS dim_customer (
                    customer_id INTEGER PRIMARY KEY,
                    customer_name TEXT,
                    region TEXT)''')

# 插入示例數(shù)據(jù)
cursor.execute("INSERT INTO dim_product (product_id, product_name, category) VALUES (1, 'Laptop', 'Electronics')")
cursor.execute("INSERT INTO dim_customer (customer_id, customer_name, region) VALUES (1, 'Alice', 'North America')")
cursor.execute("INSERT INTO fact_sales (sale_id, product_id, customer_id, sales_amount, sale_date) VALUES (1, 1, 1, 1200, '2023-09-01')")

conn.commit()

# 查詢數(shù)據(jù)
df = pd.read_sql_query("SELECT * FROM fact_sales", conn)
print(df)

conn.close()
-----------------------------------
?著作權(quán)歸作者所有：來自51CTO博客作者申公豹CTO的原創(chuàng)作品，請聯(lián)系作者獲取轉(zhuǎn)載授權(quán)，否則將追究法律責(zé)任
【數(shù)據(jù)飛輪】驅(qū)動業(yè)務(wù)增長的高效引擎 —從數(shù)據(jù)倉庫到數(shù)據(jù)中臺的技術(shù)進(jìn)化與實(shí)戰(zhàn) 
https://blog.51cto.com/u_16123336/12058658

from kafka import KafkaProducer
import json
import time

# 初始化Kafka生產(chǎn)者
producer = KafkaProducer(bootstrap_servers=['localhost:9092'], 
                         value_serializer=lambda v: json.dumps(v).encode('utf-8'))

# 模擬實(shí)時數(shù)據(jù)流
for i in range(10):
    data = {'event_id': i, 'event_time': time.time(), 'value': i * 100}
    producer.send('data_stream', value=data)
    print(f"Produced event: {data}")
    time.sleep(1)

producer.close()
-----------------------------------
?著作權(quán)歸作者所有：來自51CTO博客作者申公豹CTO的原創(chuàng)作品，請聯(lián)系作者獲取轉(zhuǎn)載授權(quán)，否則將追究法律責(zé)任
【數(shù)據(jù)飛輪】驅(qū)動業(yè)務(wù)增長的高效引擎 —從數(shù)據(jù)倉庫到數(shù)據(jù)中臺的技術(shù)進(jìn)化與實(shí)戰(zhàn) 
https://blog.51cto.com/u_16123336/12058658

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import linear_kernel
import pandas as pd

# 模擬商品數(shù)據(jù)集
products = pd.DataFrame({
    'product_id': [1, 2, 3, 4],
    'product_name': ['Laptop', 'Smartphone', 'Tablet', 'Monitor'],
    'description': ['High-performance laptop', 'Latest smartphone model', 'Affordable tablet', 'High-resolution monitor']
})

# 基于TF-IDF的推薦模型
tfidf = TfidfVectorizer(stop_words='english')
tfidf_matrix = tfidf.fit_transform(products['description'])
cosine_sim = linear_kernel(tfidf_matrix, tfidf_matrix)

# 商品推薦函數(shù)
def recommend_products(product_id, cosine_sim=cosine_sim):
    idx = products[products['product_id'] == product_id].index[0]
    sim_scores = list(enumerate(cosine_sim[idx]))
    sim_scores = sorted(sim_scores, key=lambda x: x[1], reverse=True)
    sim_scores = sim_scores[1:3]
    product_indices = [i[0] for i in sim_scores]
    return products['product_name'].iloc[product_indices]

# 假設(shè)用戶購買了Smartphone
recommended_products = recommend_products(2)
print(f"Based on your purchase, you may also like: {recommended_products.tolist()}")
-----------------------------------
?著作權(quán)歸作者所有：來自51CTO博客作者申公豹CTO的原創(chuàng)作品，請聯(lián)系作者獲取轉(zhuǎn)載授權(quán)，否則將追究法律責(zé)任
【數(shù)據(jù)飛輪】驅(qū)動業(yè)務(wù)增長的高效引擎 —從數(shù)據(jù)倉庫到數(shù)據(jù)中臺的技術(shù)進(jìn)化與實(shí)戰(zhàn) 
https://blog.51cto.com/u_16123336/12058658

Based on your purchase, you may also like: ['Laptop', 'Tablet']

import pandas as pd
from sklearn.metrics.pairwise import cosine_similarity

# 用戶評分?jǐn)?shù)據(jù)集
ratings = pd.DataFrame({
    'user_id': [1, 1, 1, 2, 2, 3, 3, 3],
    'product_id': [101, 102, 103, 101, 104, 102, 103, 104],
    'rating': [5, 4, 3, 4, 5, 2, 3, 4]
})

# 創(chuàng)建用戶-產(chǎn)品矩陣
user_product_matrix = ratings.pivot_table(index='user_id', columns='product_id', values='rating').fillna(0)

# 基于余弦相似度計算用戶相似性
user_similarity = cosine_similarity(user_product_matrix)
user_similarity_df = pd.DataFrame(user_similarity, index=user_product_matrix.index, columns=user_product_matrix.index)

# 推薦函數(shù)：基于相似用戶推薦商品
def recommend_for_user(user_id, user_product_matrix, user_similarity_df, top_n=2):
    similar_users = user_similarity_df[user_id].sort_values(ascending=False).index[1:top_n+1]
    similar_users_ratings = user_product_matrix.loc[similar_users].mean(axis=0)
    user_ratings = user_product_matrix.loc[user_id]
    recommendations = similar_users_ratings[user_ratings == 0].sort_values(ascending=False).head(top_n)
    return recommendations

# 為用戶1推薦商品
recommended_products = recommend_for_user(1, user_product_matrix, user_similarity_df)
print(f"Recommended products for user 1: {recommended_products.index.tolist()}")

Recommended products for user 1: [104]

from sklearn.cluster import KMeans
import numpy as np

# 模擬用戶生命周期數(shù)據(jù)
user_data = pd.DataFrame({
    'user_id': [1, 2, 3, 4, 5],
    'purchase_frequency': [5, 2, 10, 1, 4],
    'avg_spend': [500, 100, 1200, 50, 300],
    'last_purchase_days_ago': [10, 40, 5, 90, 20]
})

# 聚類用戶以識別生命周期階段
kmeans = KMeans(n_clusters=3, random_state=0).fit(user_data[['purchase_frequency', 'avg_spend', 'last_purchase_days_ago']])
user_data['cluster'] = kmeans.labels_

# 打印用戶的生命周期分組
print(user_data)

import pandas as pd
from statsmodels.tsa.holtwinters import ExponentialSmoothing

# 模擬歷史銷售數(shù)據(jù)
sales_data = pd.DataFrame({
    'month': pd.date_range(start='2023-01-01', periods=12, freq='M'),
    'sales': [200, 220, 240, 260, 250, 300, 320, 340, 360, 380, 400, 420]
})
sales_data.set_index('month', inplace=True)

# 使用指數(shù)平滑法進(jìn)行庫存預(yù)測
model = ExponentialSmoothing(sales_data['sales'], trend='add', seasonal=None)
model_fit = model.fit()
forecast = model_fit.forecast(steps=3)

print("Future Inventory Forecast:", forecast)

Future Inventory Forecast: 2024-01-31    439.545451
2024-02-29    459.860136
2024-03-31    480.174820
Freq: M, dtype: float64

# 啟動Kafka服務(wù)
bin/zookeeper-server-start.sh config/zookeeper.properties
bin/kafka-server-start.sh config/server.properties

# 創(chuàng)建一個新的Kafka主題
bin/kafka-topics.sh --create --topic user-behavior --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

from kafka import KafkaProducer
import json

# 連接到Kafka服務(wù)器
producer = KafkaProducer(bootstrap_servers='localhost:9092', value_serializer=lambda v: json.dumps(v).encode('utf-8'))

# 發(fā)送用戶行為數(shù)據(jù)
user_behavior = {
    'user_id': 1,
    'event': 'click',
    'item_id': 101,
    'timestamp': '2024-09-09 12:00:00'
}
producer.send('user-behavior', user_behavior)

from pyspark.sql import SparkSession

# 初始化Spark會話
spark = SparkSession.builder.appName("BatchProcessing").getOrCreate()

# 讀取CSV文件
data = spark.read.csv("user_data.csv", header=True, inferSchema=True)

# 數(shù)據(jù)處理，計算每個用戶的購買總數(shù)
total_purchase = data.groupBy("user_id").sum("purchase_amount")
total_purchase.show()

# 使用Flink處理實(shí)時數(shù)據(jù)流
env = StreamExecutionEnvironment.get_execution_environment()

# 從Kafka獲取數(shù)據(jù)流
kafka_consumer = FlinkKafkaConsumer(
    'user-behavior', 
    SimpleStringSchema(), 
    {'bootstrap.servers': 'localhost:9092'}
)
stream = env.add_source(kafka_consumer)

# 處理數(shù)據(jù)流
stream.map(lambda event: (event['user_id'], 1)) \
      .key_by(lambda x: x[0]) \
      .sum(1) \
      .print()

# 啟動Flink流處理任務(wù)
env.execute("UserBehaviorStreamProcessing")

import numpy as np
from sklearn.decomposition import NMF

# 用戶評分矩陣
R = np.array([[5, 3, 0, 1],
              [4, 0, 0, 1],
              [1, 1, 0, 5],
              [1, 0, 0, 4],
              [0, 1, 5, 4]])

# 使用非負(fù)矩陣分解（NMF）分解評分矩陣
model = NMF(n_compnotallow=2, init='random', random_state=0)
W = model.fit_transform(R)
H = model.components_

# 重新構(gòu)建評分矩陣
R_predicted = np.dot(W, H)
print("Predicted Ratings:\n", R_predicted)
-----------------------------------
?著作權(quán)歸作者所有：來自51CTO博客作者申公豹CTO的原創(chuàng)作品，請聯(lián)系作者獲取轉(zhuǎn)載授權(quán)，否則將追究法律責(zé)任
【數(shù)據(jù)飛輪】驅(qū)動業(yè)務(wù)增長的高效引擎 —從數(shù)據(jù)倉庫到數(shù)據(jù)中臺的技術(shù)進(jìn)化與實(shí)戰(zhàn) 
https://blog.51cto.com/u_16123336/12058658

Predicted Ratings:
 [[5.25583751 1.99314304 0.         1.45510614]
 [3.50429883 1.32891643 0.         0.97018348]
 [1.31291255 0.9441558  1.94957474 3.94614513]
 [0.98126695 0.72179626 1.52760301 3.0788861 ]
 [0.         0.65008539 2.83998144 5.21892451]]

# 假設(shè)我們已經(jīng)訓(xùn)練好推薦模型
def recommend_products(user_id, R_predicted, top_n=2):
    # 獲取用戶未評分的產(chǎn)品，并按預(yù)測評分排序
    user_ratings = R_predicted[user_id]
    recommendations = [(i, rating) for i, rating in enumerate(user_ratings) if rating > 0]
    recommendations.sort(key=lambda x: x[1], reverse=True)
    return recommendations[:top_n]

# 為用戶0推薦商品
user_0_recommendations = recommend_products(0, R_predicted)
print(f"Recommended products for user 0: {user_0_recommendations}")

from sklearn.linear_model import SGDRegressor

# 假設(shè)已有部分用戶行為數(shù)據(jù)
X = np.array([[1, 2], [4, 5], [7, 8]])
y = np.array([1, 2, 3])

# 使用SGD回歸模型進(jìn)行在線學(xué)習(xí)
model = SGDRegressor()

# 模擬新數(shù)據(jù)流入，并實(shí)時更新模型
for i in range(3):
    X_new = np.array([[i + 10, i + 11]])
    y_new = np.array([i + 4])
    model.partial_fit(X_new, y_new)

# 預(yù)測新數(shù)據(jù)
pred = model.predict([[15, 16]])
print("Prediction for new data:", pred)
-----------------------------------
?著作權(quán)歸作者所有：來自51CTO博客作者申公豹CTO的原創(chuàng)作品，請聯(lián)系作者獲取轉(zhuǎn)載授權(quán)，否則將追究法律責(zé)任
【數(shù)據(jù)飛輪】驅(qū)動業(yè)務(wù)增長的高效引擎 —從數(shù)據(jù)倉庫到數(shù)據(jù)中臺的技術(shù)進(jìn)化與實(shí)戰(zhàn) 
https://blog.51cto.com/u_16123336/12058658

Prediction for new data: [19.29666937]