WIP: source: HDFS #145

sudharsana-kjl · 2019-07-29T16:16:49Z

This PR hopes to add HDFS source.

lgtm-com · 2019-07-29T17:19:56Z

This pull request introduces 5 alerts when merging fc82d86 into 44744d8 - view on LGTM.com

new alerts:

5 for Unused import

lgtm-com · 2019-07-29T19:11:29Z

This pull request introduces 5 alerts when merging cd91197 into 44744d8 - view on LGTM.com

new alerts:

5 for Unused import

johnandersen777 · 2019-07-30T17:59:52Z

tests/source/test_hdfs.py

+from dffml.source.csv import CSVSource, CSVSourceConfig
+
+
+class TestHDFSSource(FileSourceTest, AsyncTestCase):


FileSourceTest should be SourceTest

johnandersen777 · 2019-07-30T18:02:55Z

dffml/source/hdfs_source.py

+    port: str
+    user: str
+    filepath: str
+    source: BaseSource.load


source: FileSource

johnandersen777 · 2019-07-30T18:18:49Z

dffml/source/hdfs_source.py

+        self.client = InsecureClient(
+            "http://" + self.config.host + ":" + self.config.port,
+            user="hadoopuser",
+        )


async def new_open(source): with self.client.read(self.config.filepath, encoding="utf-8") as fd: await source.load_fd(fd) self.config.source._open = new_open

Also, give it a new close method, which will be called when we exit from the filesource's context.

To enter the context, we usually do the async with statement:

async with self.config.source as source: # this is where new_open get called # At the end of this with block new_close would get called, but we don't want to call it until the __aexit__ method of HDFSSource

You'll need to do something like this, also make sure to modify __aexit__ similarly.

dffml/dffml/df/memory.py

Lines 780 to 792 in 927fb73

) -> "MemoryOperationImplementationNetworkContext":

self.__stack = AsyncExitStack()

await self.__stack.__aenter__()

self.operations = {

opimp.op.name: await self.__stack.enter_async_context(opimp)

for opimp in self.opimps.values()

}

return self

async def __aexit__(self, exc_type, exc_value, traceback):

if self.__stack is not None:

await self.__stack.aclose()

self.__stack = None

You'll use

self.source = await self.__stack.enter_async_context(self.config.source)

Implement __call__ method for HDFS source which does: return self.source()

dffml/dffml/source/source.py

Lines 60 to 61 in eb9b600

def __call__(self) -> BaseSourceContext:

return self.CONTEXT(self)

johnandersen777 · 2019-08-06T16:20:37Z

dffml/source/hdfs_source.py

+    @classmethod
+    def args(cls, args, *above) -> Dict[str, Arg]:
+        cls.config_set(args, above, "host", Arg(type=str))
+        cls.config_set(args, above, "port", Arg(type=str))


johnandersen777 · 2019-08-12T16:02:49Z

.ci/Dockerfile

+  mv hadoop-3.1.2 /usr/local && \
+  ln -sf /usr/local/hadoop-3.1.2/ /usr/local/hadoop
+
+CMD cat hadoop_config >> ~/.bashrc


You need to COPY hadoop_config into the container first

I've included the following line:

ADD configs/hadoop_config /usr/local/hadoop CMD cat /usr/local/hadoop/hadoop_config >> ~/.bashrc

Is this fine?

johnandersen777 · 2019-08-12T16:15:29Z

.ci/Dockerfile

+
+CMD bash cat ~/.ssh/id_rsa.pub >> ~/.ssh/authorized_keys \
+    ssh-copy-id -i ~/.ssh/id_rsa.pub localhost \
+    ssh localhost


This won't work because the ssh server isn't running. You can safely delete these steps

Oh, so there's no need to fetch the authorized keys?

Or I should just remove ssh localhost ?

Got it 👍

johnandersen777 · 2019-08-12T16:16:34Z

.ci/Dockerfile

+
+## ssh key gen
+RUN \
+    ssh-keygen -t rsa -P "" -f ~/.ssh/id_rsa


You don't need this, since you don't need to create an ssh key for the root account within the container

johnandersen777 · 2019-08-12T16:16:43Z

.ci/Dockerfile

+##    mkdir ~/.ssh \
+##    chmod 700 ~/.ssh
+
+CMD bash cat "" >> ~/.ssh/authorized_keys \


You can remove this

johnandersen777 · 2019-08-12T16:18:59Z

.ci/configs/ssh_config

@@ -0,0 +1,3 @@
+Host *
+  UserKnownHostsFile /dev/null 
+  StrictHostKeyChecking no


You should remove this file

johnandersen777 · 2019-08-12T16:19:04Z

.ci/start-hadoop.sh

@@ -0,0 +1,15 @@
+#!/bin/bash
+


You probably want to source /usr/local/hadoop/hadoop_config

- Add another dockerfile that sets up a mini cli hadoop cluster - modify new_open() for hdfs_source

Signed-off-by: John Andersen <[email protected]>

lgtm-com · 2019-08-21T16:32:52Z

This pull request introduces 5 alerts when merging 56aab8f into eb9b600 - view on LGTM.com

new alerts:

5 for Unused import

johnandersen777 · 2019-08-23T02:34:27Z

diff --git a/dffml/source/hdfs_source.py b/dffml/source/hdfs_source.py
index 9f90f62..cc2a919 100644
--- a/dffml/source/hdfs_source.py
+++ b/dffml/source/hdfs_source.py
@@ -24,12 +24,13 @@ class HDFSSourceConfig(BaseConfig, NamedTuple):
 class HDFSSource(BaseSource):
 
     CONTEXT = BaseSourceContext
+
     async def new_open(self):
         with self.client.read(self.config.filepath, encoding="utf-8") as fd:
             await self.config.source.load_fd(fd)
 
     async def new_close(self):
-        with self.client.write(self.config.filepath, encoding="utf-8") as fd:
+        with self.client.write(self.config.filepath, encoding="utf-8", overwrite=True) as fd:
             await self.config.source.dump_fd(fd)
 
     async def __aenter__(self) -> "BaseSource":
@@ -37,12 +38,29 @@ class HDFSSource(BaseSource):
             "http://" + self.config.host + ":" + self.config.port,
             user="hadoopuser",
         )
-        self.config.source._open = await self.new_open()
-        self.config.source._close = await self.new_close()
+        # This might be a problem
+        self.config.source._open = self.new_open.__get__(self.config.source, self.config.source.__class__)
+        self.config.source._close = self.new_close.__get__(self.config.source, self.config.source.__class__)
+        # Create AsyncExitStack and call it's __aenter__ method
+        self.__stack = AsyncExitStack()
+        await self.__stack.__aenter__()
+        # Enter the context of the source
+        # Same as:
+        # async with self.config.source as self.source:
+        self.source = await self.__stack.enter_async_context(self.config.source)
+        return self
+
+    async def __aexit__(self, exc_type, exc_value, traceback):
+        if self.__stack is not None:
+            await self.__stack.__aexit__(exc_type, exc_value, traceback)
+            self.__stack = None
         return self
 
     def __call__(self) -> BaseSourceContext:
-        return self.config.source()
+        # Call the source
+        # Same as:
+        # async with self.source() as source_context:
+        return self.source()
 
     @classmethod
     def args(cls, args, *above) -> Dict[str, Arg]:

You may need to add the @staticmethod decorator to the new_open and new_close methods of HDFSSource

lgtm-com · 2019-09-01T14:14:21Z

This pull request introduces 6 alerts when merging 979f83a into 2864a11 - view on LGTM.com

new alerts:

6 for Unused import

… into hadoop_source

lgtm-com · 2019-09-03T16:39:46Z

This pull request introduces 6 alerts when merging 743de45 into f4a5dda - view on LGTM.com

new alerts:

6 for Unused import

lgtm-com · 2019-09-13T12:03:07Z

This pull request introduces 7 alerts when merging 3426b39 into 6d8f8e9 - view on LGTM.com

new alerts:

6 for Unused import
1 for Non-callable called

lgtm-com · 2019-09-24T15:23:37Z

This pull request introduces 6 alerts when merging 5210a3e into c8f9ba3 - view on LGTM.com

new alerts:

6 for Unused import

lgtm-com · 2019-10-01T15:46:19Z

This pull request introduces 9 alerts when merging 0b6dfcb into d3bf6c9 - view on LGTM.com

new alerts:

8 for Unused import
1 for Except block handles 'BaseException'

lgtm-com · 2019-10-07T17:39:43Z

This pull request introduces 10 alerts when merging 95f7b73 into 65e4ce4 - view on LGTM.com

new alerts:

9 for Unused import
1 for Except block handles 'BaseException'

Signed-off-by: John Andersen <[email protected]>

lgtm-com · 2019-10-07T22:24:26Z

This pull request introduces 13 alerts when merging 99764b0 into 65e4ce4 - view on LGTM.com

new alerts:

12 for Unused import
1 for Except block handles 'BaseException'

Signed-off-by: John Andersen <[email protected]>

lgtm-com · 2019-10-08T16:58:57Z

This pull request introduces 15 alerts when merging 82d6e16 into 65e4ce4 - view on LGTM.com

new alerts:

12 for Unused import
2 for Unused local variable
1 for Except block handles 'BaseException'

hdfs connection setup

fc82d86

sudharsana-kjl added 2 commits July 29, 2019 23:05

source: hdfsSource: modify source arg

11c2c61

source: hdfs: remove test_update()

cd91197

johnandersen777 reviewed Jul 30, 2019

View reviewed changes

johnandersen777 reviewed Aug 6, 2019

View reviewed changes

sudharsana-kjl added 5 commits August 9, 2019 22:13

dockerfile for hadoop setup

10edbe0

add config files for hadoop docker setup

430284d

another approach to setup hadoop

e236c24

modify dockerfile

1f889e6

fix Dockerfile

9438b4c

johnandersen777 reviewed Aug 12, 2019

View reviewed changes

add hadoop_config to container

ba5c863

johnandersen777 reviewed Aug 12, 2019

View reviewed changes

sudharsana-kjl and others added 9 commits August 12, 2019 23:02

remove ssh configs in Dockerfile

bb4ec87

Modify Dockerfile & modify source: hdfs_source

4b78517

- Add another dockerfile that sets up a mini cli hadoop cluster - modify new_open() for hdfs_source

hdfs

5f8431b

Signed-off-by: John Andersen <[email protected]>

add usage to hdfs dockerfile

c081b7d

Signed-off-by: John Andersen <[email protected]>

hdfs connection setup

0a02560

source: hdfsSource: modify source arg

02d9792

source: hdfs: remove test_update()

a185cb4

dockerfile for hadoop setup

0bcca2f

add config files for hadoop docker setup

cd0e355

modify writestream in new_close()

56aab8f

sudharsana-kjl added 3 commits August 24, 2019 08:27

source:hdfs: add __aexit__()

4c4331b

package hdfs source

94f5c30

Merge branch 'master' into hadoop_source

979f83a

sudharsana-kjl added 2 commits September 3, 2019 21:44

modify test path for hdfs_source test

2fa1009

Merge branch 'hadoop_source' of https://github.com/sudharsana-kjl/dffml…

743de45

… into hadoop_source

sudharsana-kjl added 5 commits September 13, 2019 06:39

move from dffml/source/hdfs to source/hdfs

b52b3eb

update branch with master

32d4446

restructure source/hdfs

2789e1d

change hdfscli to hdfs

a8bd20a

remover pip-wheel-metadat

3426b39

modify hdfs test sample file path

5210a3e

sudharsana-kjl added 2 commits October 1, 2019 19:25

Merge remote-tracking branch 'upstream/master' into hadoop_source

46739d5

modify docker_hdfs set up

0b6dfcb

add sample_data.csv and modify hadoop_docker setup

95f7b73

source: hdfs: util: Wait for HDFS startup and yield IP

99764b0

Signed-off-by: John Andersen <[email protected]>

import patch and change context of new_open/close

82d6e16

Signed-off-by: John Andersen <[email protected]>

johnandersen777 added the backlog No activiy in a while, need to revisit label Dec 26, 2019

johnandersen777 deleted the branch intel:master March 12, 2022 22:22

johnandersen777 closed this Mar 12, 2022

		from dffml.source.csv import CSVSource, CSVSourceConfig


		class TestHDFSSource(FileSourceTest, AsyncTestCase):

	) -> "MemoryOperationImplementationNetworkContext":
	self.__stack = AsyncExitStack()
	await self.__stack.__aenter__()
	self.operations = {
	opimp.op.name: await self.__stack.enter_async_context(opimp)
	for opimp in self.opimps.values()
	}
	return self

	async def __aexit__(self, exc_type, exc_value, traceback):
	if self.__stack is not None:
	await self.__stack.aclose()
	self.__stack = None

	def __call__(self) -> BaseSourceContext:
	return self.CONTEXT(self)

WIP: source: HDFS #145

WIP: source: HDFS #145

Uh oh!

Conversation

sudharsana-kjl commented Jul 29, 2019

Uh oh!

lgtm-com bot commented Jul 29, 2019

Uh oh!

lgtm-com bot commented Jul 29, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

johnandersen777 Aug 20, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lgtm-com bot commented Aug 21, 2019

Uh oh!

johnandersen777 commented Aug 23, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

lgtm-com bot commented Sep 1, 2019

Uh oh!

lgtm-com bot commented Sep 3, 2019

Uh oh!

lgtm-com bot commented Sep 13, 2019

Uh oh!

lgtm-com bot commented Sep 24, 2019

Uh oh!

lgtm-com bot commented Oct 1, 2019

Uh oh!

lgtm-com bot commented Oct 7, 2019

Uh oh!

lgtm-com bot commented Oct 7, 2019

Uh oh!

lgtm-com bot commented Oct 8, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

johnandersen777 Aug 20, 2019 •

edited

Loading

johnandersen777 commented Aug 23, 2019 •

edited

Loading