如何将转换模块移入Docker运行的Memgraph？Kafka流集成及部署疑问

不需要用vi复制粘贴代码，有几种更高效的部署方式：

1. 卷挂载（推荐，适合开发阶段） #

将本地存放music.py的目录直接挂载到Memgraph容器的查询模块目录（默认路径为/usr/lib/memgraph/query_modules），本地修改代码后容器内会自动同步。

• 用docker run启动容器时执行：

  docker run -d -p 7687:7687 -v /你的本地模块目录绝对路径:/usr/lib/memgraph/query_modules memgraph/memgraph
  

• 用Docker Compose的话，在配置文件中添加卷挂载：

  services:
    memgraph:
      image: memgraph/memgraph
      ports:
        - "7687:7687"
      volumes:
        - ./本地模块相对路径:/usr/lib/memgraph/query_modules
  

挂载完成后，要么重启容器，要么在Memgraph中执行CALL mg.load_all();加载新模块。

2. 用Memgraph Lab上传 #

如果使用Memgraph Lab管理数据库，可直接上传模块：

1. 打开Memgraph Lab（默认访问地址http://localhost:3000）
2. 切换到Query Modules页面
3. 点击Upload Module，选择本地的music.py文件上传
4. 上传后执行CALL mg.load_all();使模块生效

3. Docker cp命令复制文件 #

直接把本地文件复制到运行中的Memgraph容器内：

1. 通过docker ps找到Memgraph容器的ID或名称
2. 执行复制命令：

   docker cp /本地路径/music.py <容器ID或名称>:/usr/lib/memgraph/query_modules/
   

3. 执行CALL mg.load_all();加载模块

你的转换模块代码问题 #

代码中存在变量名不一致的错误，会导致运行时异常：

• 解析消息后赋值给了movie_dict，但后续参数中却使用album_dict调用
• 字段引用album_dict["movie"]["movieId"]可能不符合消息结构（如果消息里是album字段，应改为album_dict["album"]["albumId"]）

修正后的关键代码片段：

import mgp
import json

@mgp.transformation
def rating(messages: mgp.Messages
                ) -> mgp.Record(query=str, parameters=mgp.Nullable[mgp.Map]):
    result_queries = []

    for i in range(messages.total_messages()):
        message = messages.message_at(i)
        # 统一变量名
        album_dict = json.loads(message.payload().decode('utf8'))
        result_queries.append(
            mgp.Record(
                query=("MERGE (u:User {id: $userId}) "
                        "MERGE (m:Album {id: $albumId, title: $title}) "
                        "WITH u, m "
                        "UNWIND $genres as genre "
                        "MERGE (m)-[:OF_GENRE]->(:Genre {name: genre}) "
                        "MERGE (u)-[r:RATED {rating: ToFloat($rating), timestamp: $timestamp}]->(m)"),
                parameters={
                    "userId": album_dict["userId"],
                    "albumId": album_dict["album"]["albumId"],
                    "title": album_dict["album"]["title"],
                    "genres": album_dict["album"]["genres"],
                    "rating": album_dict["rating"],
                    "timestamp": album_dict["timestamp"]}))

    return result_queries

内容的提问来源于stack exchange，提问作者KWriter