Chương 4: Neo4j Ingestion & Connector Configuration

Chương 4: Neo4j Ingestion & Connector Configuration#

Thành viên phụ trách: Võ Thành Minh Tuệ

1. Bối cảnh và Mục tiêu#

Chương này trình bày cách thức nhóm xây dựng một đường ống (pipeline) chuyên dụng để nạp tự động cấu trúc liên kết đồ thị (bao gồm Node và Edge) từ Kafka vào trực tiếp cơ sở dữ liệu đồ thị Neo4j. Nhằm đáp ứng chặt chẽ các ràng buộc của đồ án, kiến trúc bắt buộc phải tận dụng Neo4j Kafka Connector Sink, đảm bảo dữ liệu chạy thẳng từ message broker vào Neo4j mà không phát sinh giao tiếp qua tầng xử lý trung gian Spark. Hơn nữa, toàn bộ tiến trình nạp (ingestion) phải tuân thủ nghiêm ngặt nguyên lý Idempotency (Lũy đẳng) - mọi nỗ lực tái xử lý (replay) trên các node hay edge đã tồn tại tuyệt đối không được gây ra hiện tượng nhân bản (duplication) hoặc làm phình to tập dữ liệu rác trong đồ thị.

2. Cách tiếp cận và Lý do lựa chọn#

Nhóm đã đánh giá và thống nhất triển khai mô hình Native Ingestion thông qua Kafka Connect Worker tích hợp plugin chính thức neo4j-kafka-connect, dựa trên hai lợi điểm kỹ thuật vượt trội:

  1. Khả năng tối ưu thông lượng (Throughput Optimization): sự kiện (node/edge) được nạp streaming với độ trễ cực thấp do loại bỏ hoàn toàn chi phí khởi tạo, vận hành và giao tiếp mạng của một cụm Spark bổ sung.

  2. Khai thác sức mạnh của Cypher MERGE: connector cho phép linh hoạt nhúng và thực thi các khối lệnh Cypher chuyên biệt cho từng topic được subscribe. Việc sử dụng mệnh đề MERGE thay vì CREATE chính là chìa khóa thuật toán để đảm bảo đồ thị hội tụ về trạng thái lũy đẳng.

Cấu hình kịch bản Cypher hợp nhất đồ thị (kafka/neo4j-sink-config.json):#

1. Logic xử lý khối Node cú pháp (cpg_node_events):#

MERGE (n:Node {node_id: event.node_id})
WITH n, event, datetime(event.event_time) AS new_time
WHERE n.event_time IS NULL OR new_time > datetime(n.event_time)
SET n.event_time = event.event_time,
    n.file_path = event.file_path,
    n.label = event.label,
    n.name = event.name,
    n.line = event.line,
    n.column = event.column
SET n += coalesce(event.properties, {})

2. Logic xử lý khối Relationship định tuyến (cpg_edge_events):#

MERGE (source:Node {node_id: event.source_id}) 
MERGE (target:Node {node_id: event.target_id}) 
WITH source, target, event 
CALL apoc.merge.relationship(source, event.type, {edge_id: event.edge_id}, {}, target, {}) YIELD rel 
WITH rel, event, datetime(event.event_time) AS new_time 
WHERE rel.event_time IS NULL OR new_time > datetime(rel.event_time) 
SET rel.event_time = event.event_time, rel.file_path = event.file_path 
SET rel += coalesce(event.properties, {})

3. Mã nguồn & Kết quả thực thi#

Connector được hệ thống cấu hình nạp (provision) tự động vào vùng nhớ của container kafka-connect ngay trong chu trình khởi động của Docker Compose.

Cụ thể, script khởi tạo của Docker Compose thực thi lệnh curl POST sau để đăng ký connector với Kafka Connect Worker:

curl -s -X POST http://localhost:8083/connectors `
  -H "Content-Type: application/json" `
  -d (Get-Content kafka/neo4j-sink-config.json -Raw)

Kết quả khi đăng ký thành công:

{
  "name": "neo4j-sink-connector",
  "config": {
    "connector.class": "streams.kafka.connect.sink.Neo4jSinkConnector",
    "topics": "cpg_node_events,cpg_edge_events"
  },
  "tasks": [],
  "type": "sink"
}

Lưu ý kiến trúc: Luồng dữ liệu Node/Edge đi thẳng từ Kafka Connector Worker vào Neo4j mà không đi qua bất kỳ tầng Spark nào. Đây là điểm khác biệt quan trọng so với pipeline metadata: Neo4j nhận dữ liệu đồ thị qua Kafka Connect (zero-latency), trong khi MongoDB nhận siêu dữ liệu qua Spark Structured Streaming (micro-batch).

Giám sát Health Check của Connector thông qua giao thức REST API:

curl -s http://localhost:8083/connectors/neo4j-sink-connector/status

Kết quả thẩm định dữ liệu bằng Cypher Query trên Neo4j Browser:

  1. Truy vấn thống kê tổng dung lượng Node:

    MATCH (n:Node) RETURN count(n) AS total_nodes

    Kết quả thực tế: total_nodes: 25253

  2. Truy vấn phân tích độ trù mật của Edge phân theo loại:

    MATCH ()-[r]->() RETURN count(r) AS total_edges, type(r) AS edge_type ORDER BY edge_type

    Kết quả thực tế: Hệ thống trả về bảng phân bố chi tiết cho từng loại liên kết (ví dụ: AST_CHILD: 9063, CALLS: 2027, ASSIGNS: 1767, CONTROL_CONDITION: 383…).

  3. Kiểm tra Duplicate Node (xác minh không có node_id trùng):

    MATCH (n:Node)
WITH n.node_id AS nid, count(n) AS cnt
WHERE cnt > 1
RETURN count(nid) AS duplicate_nodes

    Kết quả thực tế: duplicate_nodes: 0 - khẳng định tuyệt đối không có node trùng

  4. Kiểm tra Duplicate Edge (xác minh không có edge_id trùng):

    MATCH ()-[r]->()
WITH r.edge_id AS eid, count(r) AS cnt
WHERE cnt > 1
RETURN count(eid) AS duplicate_edges

    Kết quả thực tế: duplicate_edges: 0 - xác nhận apoc.merge.relationship hoạt động chính xác

4. Minh chứng thực tế#

Minh chứng 1: Mô hình hóa trực quan Đồ thị Mã nguồn qua giao diện Neo4j Browser (http://localhost:7474)#

Đồ thị CPG trực quan Neo4j

Hình 4.1: Render một góc đồ thị CPG với giới hạn 50 relationships. Hình ảnh hiển thị rõ nét 24 nodes cùng các vector quan hệ đa hướng đặc trưng như DATA_DEPENDENCY, READS, AST_CHILD và CALLS.

Minh chứng 2: Xác minh tính toàn vẹn của cấu trúc Node (total_nodes)#

Kết quả đếm Node Neo4j

Hình 4.2: Cypher query xác thực số lượng node tổng cộng được nạp thành công vào hệ thống là 25.253 nodes.

Minh chứng 3: Khảo sát độ phức tạp của mạng lưới Edge (total_edges)#

Kết quả đếm Edge Neo4j

Hình 4.3: Cypher query phân tách rành mạch khối lượng các cạnh liên kết, cho thấy sự chiếm ưu thế của cấu trúc AST (AST_CHILD).

Minh chứng 4: Kết quả Duplicate Check#

Duplicate Check Neo4j

Hình 4.4: Cypher query xác nhận hệ số duplicate_nodes = 0, chứng minh tính Idempotent hoạt động hoàn hảo trong toàn bộ đồ thị.

5. Tự nhận xét và Đúc kết#

  • Tính Idempotent (Chống trùng lặp): Kiến trúc đảm bảo tính Idempotent rất chặt chẽ bằng cách sử dụng mệnh đề MERGE thay vì CREATE trong Cypher, kết hợp với các logic so sánh thời gian (WHERE n.event_time IS NULL OR new_time > datetime(n.event_time)). Thiết kế này giải quyết triệt để bài toán khi người dùng chạy lại (replay) cùng một file code nhiều lần: các node/edge không bị tạo mới trùng lặp, mà hệ thống chỉ đối chiếu và cập nhật trạng thái mới nhất nếu cần thiết.

  • Hiệu năng (Performance): Tốc độ insert dữ liệu qua lệnh MERGE chạy rời rạc (per-message) là chấp nhận được đối với lượng dữ liệu nhỏ hiện tại. Tuy nhiên, khi quy mô tăng lên hàng ngàn nodes từ các AST phức tạp, kiến trúc này bắt đầu bộc lộ độ trễ lớn và sinh ra các bottleneck.

6. Action plan#

Nếu được mở rộng dự án lên quy mô lớn (hàng ngàn file code, dung lượng lớn), cấu trúc pipeline sẽ được tinh chỉnh như sau:

  1. Mở rộng Topic (Scaling Kafka): Tăng số lượng partitions cho các topic lớn. Để giải quyết bài toán “đảm bảo thứ tự”, sẽ áp dụng cơ chế Partition Key (ví dụ: dùng tên file code làm Key). Như vậy, các message của cùng một file vẫn vào chung một partition (giữ nguyên thứ tự), đồng thời có thể chạy song song nhiều consumer cho các file khác nhau.

  2. Tối ưu Insert (Batching): Thay thế việc thực thi MERGE từng dòng đơn lẻ bằng cơ chế gom nhóm (Batching). Cụ thể, sẽ cấu hình connector hoặc tuỳ chỉnh truy vấn Cypher sử dụng kỹ thuật UNWIND, hoặc sử dụng thủ tục apoc.periodic.iterate của Neo4j nhằm đẩy nhanh tốc độ throughput (chèn nhiều bản ghi cùng một lúc).

7. Hướng dẫn chạy lại#

  1. Giám sát trạng thái hoạt động của các service lõi:

    docker compose ps neo4j kafka-connect

  2. Mở giao diện tương tác Neo4j Browser tại địa chỉ: http://localhost:7474

    • Tài khoản: neo4j

    • Mật khẩu: password

  3. Thực thi query kiểm thử để kết xuất trực quan không gian đồ thị:

    MATCH (n:Node)-[r]->(m:Node) RETURN n, r, m LIMIT 50
Contents