# DCLib_Driver

**Repository Path**: ZhenDu_ICT/DCLib_Driver

## Basic Information

- **Project Name**: DCLib_Driver
- **Description**: No description available
- **Primary Language**: Python
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2020-06-18
- **Last Updated**: 2020-12-17

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# 编程模型草稿

将本地文件在Hadoop中利用mapreduce做单词计数


首先先确定一个被递归调用的OODA，这个OODA将一个完整的本地文件进行分割

内容读入，data是一个map，这个map只有一对kv，key是文件名，value是完整的文件内容

```
data = DCLib.Map.Initfrom("file://nihao")
result = DCLib.Map.InitNone
```

初始化一个任务（Task），一个任务由三个Step构成：Observe+OrientDecide+Act，这个任务用来按照回车分割数据，形成具有多个kv的map

```
split_task = DCLib.Task.Init()
```

确定任务的观察阶段要执行的任务，传入参数是一个task，在这个OODA中不进行打标，所以是空的

```
DCLib.Task.Observe.setStrategy(split_task,None)
```

确定串行基本算子和事件算子，这里是库的导入，防止别名

```
DCLib.Task.OrientDecide.RegisterSeqOp(split_task, "seq_lib1.ppy", "function1")
DCLib.Task.OrientDecide.RegisterSeqOp(split_task, "seq_lib1.ppy", "function2")
DCLib.Task.OrientDecide.RegisterParalOp(split_task ,"paral_lib1.ppy", "kernal")
```

确定执行方式，不进行过滤，针对data进行执行，输出为splited_data

```
DCLib.Task.Act.setFilter(split_task,None)
DCLib.Task.Act.build(split_task,"kernal")
```

这里就成功设置好一个task，但是没有真正运行，一个task只有被其他的task调用，在最上层被DCLib.run，才被真正运行

初始化一个任务，执行filter程序

```
filter_task = DCLib.Task.Init()
```

不打标，假设到达的内容都是符合要求的

```
DCLib.Observe.setStrategy(filter_task, None)
```

确定串行基本算子和事件算子，这里是库的导入，防止别名

```
DCLib.Task.OrientDecide.RegisterSeqOp(filter_task, "seq_lib1.ppy", "function3")
DCLib.Task.OrientDecide.RegisterSeqOp(filter_task, "seq_lib1.ppy", "function4")
DCLib.Task.OrientDecide.RegisterParalOp(filter_task ,"paral_lib1.ppy", "kernal2")
```

任务的核函数和过滤器

```
DCLib.Task.Act.setFilter(filter_task,None)
DCLib.Task.Act.build(filter_task,"kernal2")
```



最后组成wordcount程序

```
word_count_task = DCLib.Task.Init()

DCLib.Task.Observe.setStrategy(word_count_task, split_task)

DCLib.Task.OrientDecide.RegisterSeqOp(word_count_task, "seq_lib1.ppy", "function5")
DCLib.Task.OrientDecide.RegisterSeqOp(word_count_task, "seq_lib1.ppy", "function6")
DCLib.Task.OrientDecide.RegisterParalOp(word_count_task ,"paral_lib1.ppy", "kernal3")
```

任务的核函数和过滤器

```
DCLib.Task.Act.setFilter(word_count_task,filter_task)
DCLib.Task.Act.build(word_count_task,"kernal3")
```

现在启动最外层任务

```
DCLib.run(word_count_task, data, result)
```