// 1) 连上任意一个 Mongos（K8s 里就是 mongodb 的 mongos Service）
// 2) 设置 Balancer 的活跃窗口：只在每天 23:00 到次日 06:00 搬迁
use config;

db.settings.updateOne(
  { _id: "balancer" },
  { $set: { activeWindow: { start: "23:00", stop: "06:00" } } },
  { upsert: true }   // 没有就插入，有就更新
);

// 3) 确认一下当前 Balancer 状态
sh.isBalancerRunning();          // false 表示当前不在窗口内、已停摆
db.settings.findOne({ _id: "balancer" });

# mongo-balancer-stop.yaml —— 大促前夜 20:00 强制停 Balancer
apiVersion: batch/v1
kind: CronJob
metadata:
  name: mongo-balancer-stop
  namespace: mongodb
spec:
  schedule: "0 20 10 11 *"   # 11 月 10 日 20:00（双十一前一天），按需改
  jobTemplate:
    spec:
      template:
        spec:
          restartPolicy: OnFailure
          containers:
            - name: mongo-cli
              image: mongo:6.0
              # 通过 mongosh 连 mongos Service，执行 stopBalancer
              command:
                - /bin/sh
                - -c
                - |
                  mongosh "mongodb://mongos.mongodb.svc.cluster.local:27017/admin" \
                    --eval 'sh.stopBalancer(); print("balancer stopped")'                  

import com.mongodb.ClientSessionOptions;
import com.mongodb.TransactionOptions;
import com.mongodb.client.ClientSession;
import com.mongodb.client.MongoClient;
import com.mongodb.client.MongoClients;
import com.mongodb.client.MongoCollection;
import org.bson.Document;
import java.util.concurrent.TimeUnit;

public class ShardedTxnExample {

    public void transfer(MongoClient client, String from, String to, double amount) {
        // 1) 事务选项：快照读 + 多数派写 + 提交最多 3 秒
        TransactionOptions txnOpts = TransactionOptions.builder()
                .readConcern(com.mongodb.ReadConcern.SNAPSHOT)   // 快照隔离，避免脏读
                .writeConcern(com.mongodb.WriteConcern.MAJORITY) // 多数派确认，防丢
                .maxCommitTime(3, TimeUnit.SECONDS)              // 关键：提交超 3 秒直接失败
                .build();

        // 2) 开会话、起事务
        try (ClientSession session = client.startSession()) {
            session.startTransaction(txnOpts);

            MongoCollection<Document> accounts =
                    client.getDatabase("shop").getCollection("accounts");

            // 3) 两个分片上的写，都在同一个 session 里
            accounts.updateOne(session,
                    new Document("_id", from),
                    new Document("$inc", new Document("balance", -amount)));
            accounts.updateOne(session,
                    new Document("_id", to),
                    new Document("$inc", new Document("balance", amount)));

            // 4) 提交；若 3 秒内没完成，驱动抛 MongoTransactionException
            session.commitTransaction();
        } catch (Exception e) {
            // 5) 任何异常都中止，保证原子性
            // 注意：session 用 try-with-resources 关闭，abort 由驱动在关闭时兜底
            System.err.println("事务失败，已回滚: " + e.getMessage());
            throw e;
        }
    }
}

// 1) 找出所有被标记为 jumbo 的箱子
use config;
db.chunks.find({ jumbo: true }).pretty();

// 2) 手动拆分：假设 shard key 是 { orderId: 1 }，在指定值处劈开
//    splitAt 在「精确等于该值」的边界切一刀
sh.splitAt("shop.orders", { orderId: 500000 });

// 如果是范围型热点，用 splitFind 自动在块内找中点切
sh.splitFind("shop.orders", { orderId: 750000 }).split();

// 3) 清掉 jumbo 标记（拆完后箱子变小，标记就没必要了）
db.chunks.updateOne(
  { ns: "shop.orders", min: { orderId: 500000 }, jumbo: true },
  { $unset: { jumbo: "" } }
);

// 4) 确认 Balancer 在窗口内会重新均衡
sh.startBalancer();
sh.status();   // 观察 chunks 在各 shard 的分布是否趋近均匀

# aerospike-namespace.yaml —— 通过 ConfigMap 注入 namespace 配置
apiVersion: v1
kind: ConfigMap
metadata:
  name: aerospike-conf
  namespace: aerospike
data:
  aerospike.conf: |
    namespace ads {
      # 内存：主索引 32GB + 20% 余量 ≈ 38GB；data-in-memory 关闭，数据走 SSD
      memory-size 38G
      # 复制因子 2：每个对象在两节点各有副本
      replication-factor 2
      # SSD 存储引擎
      storage-engine device {
        device /dev/nvme0n1        # K8s 里用 PVC/裸盘挂载
        filesize 220G              # 单文件略大于数据量，留碎片空间
        write-block-size 1M        # 1MB 写块，降低写放大
        data-in-memory false       # 数据本体在 SSD，仅索引在内存
        # 内存与 SSD 比例 ≈ 38G : 220G，约 1:6，符合「索引在内存、数据在盘」
      }
    }    

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("audience-pack") \
    .config("spark.aerospike.namespace", "ads") \
    .config("spark.aerospike.seedhosts", "aerospike.aerospike.svc.cluster.local:3000") \
    .config("spark.aerospike.port", "3000") \
    .getOrCreate()

# 1) 把 users 这个 set 读成 DataFrame（依赖 aerospike-spark-connector 包）
users = spark.read.format("aerospike") \
    .option("aerospike.set", "users") \
    .load()

# 2) 注册成临时视图，用标准 Spark SQL 圈人
users.createOrReplaceTempView("users")

# 3) 二级索引生效点：city / age / interest 上已建好 SI，过滤走索引不扫全表
audience = spark.sql("""
    SELECT user_id
    FROM   users
    WHERE  city    = 'Beijing'
      AND  age BETWEEN 25 AND 35
      AND  array_contains(interest, 'game')
""")

# 4) 把人群包写回 Aerospike 的另一个 set，供投放系统读取
audience.write.format("aerospike") \
    .option("aerospike.set", "pack_game_beijing") \
    .mode("overwrite") \
    .save()

spark.stop()

# 在 ads 命名空间的 users set 上，给 city 建字符串二级索引
asadm -e "manage indexes create string city on ads.users"

# aerospike-xdr.yaml —— A 机房的 XDR 配置（B 机房镜像对称配置）
apiVersion: v1
kind: ConfigMap
metadata:
  name: aerospike-xdr-conf
  namespace: aerospike
data:
  aerospike.conf: |
    namespace ads {
      memory-size 38G
      replication-factor 2
      storage-engine device {
        device /dev/nvme0n1
        filesize 220G
        write-block-size 1M
      }
      # XDR 开启：把本 namespace 的写同步到远端 DC
      xdr {
        enable-xdr true
        # 声明远端 B 机房为一个 datacenter
        datacenter dc-b {
          seed-nodes 10.20.0.10 10.20.0.11  # B 机房节点 IP
          # 关键：默认只 ship 本地写；来自 dc-b 的写不会回 ship，避免回环
          ship-only-local-writes true
        }
      }
    }